### ----------------------------------------------------
### -- Protocol::Postgres
### -- Licenses: Artistic-2.0
### -- Authors: Leon Timmermans
### -- File: lib/Protocol/Postgres.rakumod
### ----------------------------------------------------
### -- Chunk 2 of 3
	method postgres-type-for(TypeMap --> Type) {
		return Captured;
	}
}
my role As::Indirect[::Captured] does As {
	method postgres-type-for(TypeMap $typemap --> Type) {
		return $typemap.for-type(Captured);
	}
}

my class Type::Wrap does Type {
	has Int:D $.oid is required;
	has Type $.inner is required handles <type-object encode-to-text decode-from-text>
}
my role As::Oid[Int $oid] does As {
	method postgres-type-for(TypeMap $typemap --> Type) {
		my $inner = $typemap.for-oid($oid);
		$inner.oid === $oid ?? $inner !! Type::Wrap.new(:$oid, :$inner);
	}
}

our proto typed-as(|) is export(:typed-as) { * }

multi typed-as(Any $value, Type $type) {
	$value but As::Direct[$type];
}
multi typed-as(Any $value, Any:U $type) {
	$value but As::Indirect[$type];
}
multi typed-as(Any $value, Int $oid) {
	$value but As::Oid[$oid];
}

role TypeMap {
	proto method for-type(Any --> Type) { * }
	multi method for-type(As $value --> Type) {
		$value.postgres-type-for(self);
	}
	method for-types(@types) {
		@types.map: { self.for-type($^type) };
	}

	method for-oid(Int --> Type) { ... }
	method for-oids(@oids) {
		@oids.map: { self.for-oid($^oid) };
	}

	multi method encode(Any $input, Type :$type!, Format :$format = Text) {
		$type.encode($format, $input);
	}
	multi method encode(Any $input, Int:D :$oid!, Format :$format = Text) {
		self.for-oid($oid).encode($format, $input);
	}
	multi method encode(Any $input, Any:U :$type = $input.WHAT, Format :$format = Text) {
		self.for-type($type).encode($format, $input);
	}

	multi method decode(Blob $input, Type :$type!, Format :$format = Text) {
		$type.decode($format, $input);
	}
	multi method decode(Blob $input, Any:U :$type!, Format :$format = Text) {
		self.for-type($type).decode($format, $input);
	}
	multi method decode(Blob $input, Int:D :$oid!, Format :$format = Text) {
		self.for-oid($oid).decode($format, $input);
	}
}

class TypeMap::Minimal does TypeMap {
	multi method for-type(Any --> Type) { Type::Str }
	method for-oid(Int --> Type) { Type::Str }
}

role TypeMap::Core does TypeMap {
	multi method for-type(Cool) { Type::Str }
	multi method for-type(Blob) { Type::Blob }
	multi method for-type(Bool) { Type::Bool }
	multi method for-type(Int) { Type::Int }
	multi method for-type(Num) { Type::Num }
	multi method for-type(DateTime) { Type::DateTime }
	multi method for-type(Date) { Type::Date }
	multi method for-type(Rat) { Type::Rat }

	multi method for-type(Array[Bool] $array) { Type::Array[Type::Bool, 1000] }
	multi method for-type(Array[Blob] $array) { Type::Array[Type::Blob, 1001] }
	multi method for-type(Array[Int] $array) { Type::Array[Type::Int, 1016] }
	multi method for-type(Array[Num] $array) { Type::Array[Type::Num, 1022] }
	multi method for-type(Array[Date] $array) { Type::Array[Type::Date, 1082] }
	multi method for-type(Array[DateTime] $array) { Type::Array[Type::DateTime, 1085] }

	multi method for-oid(Int) { Type::Str }
	multi method for-oid(16) { Type::Bool }
	multi method for-oid(17) { Type::Blob }
	multi method for-oid(Int $ where 20|21|23|26) { Type::Int }
	multi method for-oid(Int $ where 700|701) { Type::Num }
	multi method for-oid(1082) { Type::Date }
	multi method for-oid(Int $ where 1114|1184) { Type::DateTime }
	multi method for-oid(1700) { Type::Rat }

	multi method for-oid(1000) { Type::Array[Type::Bool, 1000] }
	multi method for-oid(1001) { Type::Array[Type::Blob, 1001] }
	multi method for-oid(Int $ where 1002|1003|1009|1014|1015) { Type::Array[Type::Str, 1009] }
	multi method for-oid(Int $ where 1005|1007|1016|1028) { Type::Array[Type::Int, 1016] }
	multi method for-oid(Int $ where 1021|1022) { Type::Array[Type::Num, 1022] }
	multi method for-oid(1182) { Type::Array[Type::Date, 1182] }
	multi method for-oid(Int $ where 1115|1185) { Type::Array[Type::DateTime, 1185] }
	multi method for-oid(1231) { Type::Array[Type::Rat, 1231] }
	multi method for-oid(Int $ where 114|3802) { Type::JSON }
}

class TypeMap::Native does TypeMap::Core {
	multi method for-type(List $array) { Type::Array[Type::Str, 0] }
}

class TypeMap::JSON does TypeMap::Core {
	multi method for-type(List) { Type::JSON }
	multi method for-type(Map) { Type::JSON }
}

my role TypeMap::Overlay {
	has Type %!for-type{Any:U};
	has Type %!for-oid{Int};

	method add-type(Type $type, Any:U :$type-object = $type.type-object, Int:D :$oid = $type.oid --> Type) {
		%!for-type{$type-object} = $type;
		%!for-oid{$oid} = $type;
		$type;
	}

	method for-type(Any $type --> Type) {
		my $what = $type.WHAT;
		%!for-type{$what}:exists ?? %!for-type{$what} !! nextsame;
	}

	method for-oid(Int $oid --> Type) {
		%!for-oid{$oid}:exists ?? %!for-oid{$oid} !! nextsame;
	}
}

my role Protocol {
	proto method incoming-message(Packet::Base $packet) { * }
	multi method incoming-message(Packet::NoticeResponse $) {}
	method finished() {}
	method !failed-description(--> Str) { ... }
	method failed-client(Str $cause) {
		my $desc = self!failed-description ~ ': ' ~ $cause;
		self.failed-exception(X::Client.new($desc));
	}
	method failed-server(%values) {
		my $e = X::Server.new(self!failed-description, %values);
		self.failed-exception($e);
	}
	method failed-exception(Exception $e) { ... }
}

role Authenticator {
	proto method incoming-message(Packet::Authentication $packet) { * }
	multi method incoming-message(Packet::Authentication $packet) {
		die X::Client.new('Unknown authentication method');
	}
}

my class Authenticator::Null does Authenticator {
	multi method incoming-message(Packet::Authentication $packet) {
		die X::Client.new('Password required but not given');
	}
}

my class Authenticator::Password does Authenticator {
	has Str:D $.user is required;
	has Str:D $.password is required;
	has $!scram;

	multi method incoming-message(Packet::AuthenticationCleartextPassword $) {
		Packet::PasswordMessage.new(:$!password);
	}

	multi method incoming-message(Packet::AuthenticationMD5Password $packet) {
		require OpenSSL::Digest <&md5>;
		die X::Client.new('Could not load MD5 module') unless &md5;
		my sub md5-hex(Str $input) { md5($input.encode('latin1')).list».fmt('%02x').join };
		my $first-hash = md5-hex($!password ~ $!user);
		my $second-hash = md5-hex($first-hash ~ $packet.salt.decode('latin1'));
		my $password = 'md5' ~ $second-hash;
		Packet::PasswordMessage.new(:$password);
	}

	multi method incoming-message(Packet::AuthenticationSASL $packet) {
		my @mechanisms = $packet.mechanisms;
		die X::Client.new("Client does not support SASL mechanisms: @mechanisms[]") if none(@mechanisms) eq 'SCRAM-SHA-256';
		require Auth::SCRAM::Async;
		my $class = ::('Auth::SCRAM::Async::Client');
		die X::Client.new('Could not load SCRAM module') if $class === Any;
		$!scram = $class.new(:username($!user), :$!password, :digest(::('Auth::SCRAM::Async::SHA256')));
		my $initial-response = $!scram.first-message.encode;
		Packet::SASLInitialResponse.new(:mechanism<SCRAM-SHA-256>, :$initial-response);
	}

	multi method incoming-message(Packet::AuthenticationSASLContinue $packet) {
		my $client-payload = $!scram.final-message($packet.server-payload.decode).encode;
		Packet::SASLResponse.new(:$client-payload);
	}

	multi method incoming-message(Packet::AuthenticationSASLFinal $packet) {
		if not $!scram.validate($packet.server-payload.decode) {
			die X::Client.new('SCRAM final server message did not verify');
		}
	}
}

my class Protocol::Authenticating does Protocol {
	has Authenticator:D $.authenticator is required;
	has Promise $.startup-promise is required handles(:failed-exception<break>);
	has &.send-messages is required;

	multi method incoming-message(Packet::Authentication $authentication) {
		with $!authenticator.incoming-message($authentication) -> $packet {
			&!send-messages([ $packet ]);
		}
		CATCH {
			when X::Client { $!startup-promise.break($_) }
			when Str { $!startup-promise.break(X::Client.new(~$_)) }
		}
	}

	method finished() {
		$!startup-promise.keep unless $!startup-promise
	}
	method !failed-description(--> 'Could not authenticate') {}
}

multi compress-formats(@formats where all(@formats) === Text) is default { () }
multi compress-formats(@formats where all(@formats) === Binary) { (Binary) }
multi compress-formats(@formats) { @formats }

sub parameter-is-named(Parameter $param) {
	$param.named && !$param.slurpy;
}
my $mu-candidates = any(Mu.can("new")[0].candidates);
sub code-has-named(Code $candidate) {
	$candidate === $mu-candidates || parameter-is-named(any($candidate.signature.params));
}
sub constructor-is-positional(::Type) {
	so code-has-named(none(Type.can('new')[0].candidates));
}

class ResultSet {
	has Str @.columns is required;
	has Supply $.rows is required;

	method hash-rows() {
		$!rows.map(-> @row { hash @!columns Z=> @row });
	}
	method object-rows(::Class, Bool :$positional = constructor-is-positional(Class)) {
		if $positional {
			$!rows.map: -> @row { Class.new(|@row) };
		} else {
			self.hash-rows.map: -> %hash { Class.new(|%hash) };
		}
	}

	method arrays() { self.rows.list }
	method array()  { await self.rows.first }

	method value()  { self.array.head }

	method hashes() { self.hash-rows.list }
	method hash()   { await self.hash-rows.first }

	method objects(::Class, Bool :$positional = constructor-is-positional(Class)) { self.object-rows(Class, :$positional).list }
	method object(::Class, Bool :$positional = constructor-is-positional(Class)) { await self.object-rows(Class, :$positional).first }

	class Source { ... }
	class Decoder {
		has Str @.names;
		has Type @.types;
		has Format @.formats;
		has Format @.compressed-formats;

		method new(TypeMap $typemap, FieldDescription @fields, Bool $override = False) {
			my @names = @fields».name;
			my @types = $typemap.for-oids(@fields».type);
			my @formats = $override ?? @types».format !! @fields».format;
			my @compressed-formats = compress-formats(@formats);
			self.bless(:@names, :@types, :@formats, :@compressed-formats);
		}
		method decode(@row) {
			my @result;
			for zip(@!formats, @!types, @row) -> ($format, $type, $value) {
				@result.push: $type.decode($format, $value);
			}
			@result;
		}
		method make-source {
			Source.new(:decoder(self));
		}
	}

	class Source {
		has Decoder:D $.decoder is required;
		has Supplier:D $!rows handles<done quit> = Supplier::Preserving.new;
		method add-row(@row) {
			$!rows.emit($!decoder.decode(@row));
		}
		method resultset() {
			my @columns = $!decoder.names;
			ResultSet.new(:@columns, :rows($!rows.Supply));
		}
	}
}

my class Protocol::Query does Protocol {
	has TypeMap $.typemap is required;
	has ResultSet::Source $!source;
	has Supplier:D $.supplier handles(:finished<done>, :failed-exception<quit>) is required;

	multi method incoming-message(Packet::RowDescription $row) {
		my $decoder = ResultSet::Decoder.new($!typemap, $row.fields);
		$!source = $decoder.make-source;
		$!supplier.emit($!source.resultset);
	}
	multi method incoming-message(Packet::DataRow $row) {
		$!source.add-row($row.values);
	}
	multi method incoming-message(Packet::EmptyQueryResponse $) {
		my $source = ResultSet::Decoder.new($!typemap, Array[FieldDescription].new).make-source;
		$!supplier.emit($source.resultset);
	}
	multi method incoming-message(Packet::CommandComplete $) {
		$!source.done with $!source;
		$!source = Nil;
	}
	method !failed-description(--> 'Could not query') {}
}

my enum Stage (:Parsing<parse>, :Binding<bind>, :Describing<describe>, :Executing<execute>, :CopyingFrom('copy from'), :CopyingTo('copying to'), :Closing<close>, :Syncing<sync>);

my role Protocol::ExtendedQuery does Protocol {
	has Promise:D $.result is required;
	has ResultSet::Source $.source;
	has Stage:D $.stage = Parsing;
	has Supplier $!copy-from;
	has &.send-messages is required;
	multi method incoming-message(Packet::DataRow $row) {
		$!source.add-row($row.values);
	}
	multi method incoming-message(Packet::NoData $) {
		$!stage = Executing;
	}
	multi method incoming-message(Packet::EmptyQueryResponse $) {
		$!stage = Closing;
		$!result.keep('EMPTY');
	}
	multi method incoming-message(Packet::CopyInResponse $packet) {
		$!stage = CopyingTo;
		my $supplier = Supplier.new;
		sub tap($row) {
			&!send-messages([ Packet::CopyData.new(:$row) ]);
		}
		sub done() {
			&!send-messages([ Packet::CopyDone.new, Packet::Sync.new ]);
		}
		sub quit($reason) {
			&!send-messages([ Packet::CopyFail.new(~$reason), Packet::Sync.new ]);
		}
		my $supply = $supplier.Supply;
		$supply = $supply.map(*.encode) if $packet.format === Text;
		$supply.act(&tap, :&done, :&quit);
		$!result.keep($supplier);
	}
	multi method incoming-message(Packet::CopyOutResponse $packet) {
		$!stage = CopyingFrom;
		$!copy-from = Supplier::Preserving.new;
		my $supply = $!copy-from.Supply;
		$supply = $supply.map(*.decode) if $packet.format === Text;
		$!result.keep($supply);
	}
	multi method incoming-message(Packet::CopyData $packet) {
		$!copy-from.emit($packet.row);
	}
	multi method incoming-message(Packet::CopyDone $) {
		$!copy-from.done;
	}
	multi method incoming-message(Packet::CopyFail $packet) {
		$!copy-from.quit($packet.reason);
	}
	multi method incoming-message(Packet::CommandComplete $packet) {
		$!stage = Closing;
		if $!source {
			$!source.done;
		} elsif not $!result {
			$!result.keep($packet.tag);
		}
	}
	multi method incoming-message(Packet::CloseComplete $) {
		$!stage = Syncing;
	}
	method !failed-description() {
		"Could not $!stage.value()";
	}
	method failed-exception(Exception $exception) {
		if not $!result {
			$!result.break($exception);
		} elsif $!source {
			$!source.quit($exception);
		}
	}
}

my class Protocol::BindingQuery does Protocol::ExtendedQuery {
	has TypeMap $.typemap is required;
	multi method incoming-message(Packet::ParseComplete $) {
		$!stage = Binding;
	}
	multi method incoming-message(Packet::BindComplete $) {
		$!stage = Describing;
	}
	multi method incoming-message(Packet::RowDescription $packet) {
		$!stage = Executing;
		my $decoder = ResultSet::Decoder.new($!typemap, $packet.fields);
		$!source = $decoder.make-source;
		$!result.keep($!source.resultset);
	}
}

class Client { ... }

class PreparedStatement {
	has Client:D $!client is built is required;
	has Str:D $!name is built is required;
	has Type @!input-types is built is required;
	has Format @!input-formats = compress-formats(@!input-types».format);
	has ResultSet::Decoder $!decoder is built is required;
	has Bool $!closed = False;
	method execute(@values?) {
		die X::Client.new('Prepared statement already closed') if $!closed;
		die X::Client.new("Wrong number or arguments, got {+@values} expected {+@!input-types}") if @values != @!input-types;

		my @fields = @!input-types Z[&type-encode] @values;
		my @result-formats = $!decoder ?? $!decoder.compressed-formats !! ();
		my $bind = Packet::Bind.new(:$!name, :formats(@!input-formats), :@fields, :@result-formats);

		$!client.execute-prepared($bind, $!decoder);
	}
	method close(--> Promise) {
		$!closed = True;
		$!client.close-prepared($!name);
	}
	method DESTROY() {
		self.close unless $!closed;
	}
}

my class Protocol::Prepare does Protocol {
	has Client:D $.client is required;
	has Str:D $.name is required;
	has Promise:D $.result is required handles(:failed-exception<break>);
	has Type @!input-types;
	has ResultSet::Decoder $!decoder;

	multi method incoming-message(Packet::ParseComplete $) {
	}
	multi method incoming-message(Packet::ParameterDescription $packet) {
		@!input-types = $!client.typemap.for-oids($packet.types);
	}
	multi method incoming-message(Packet::RowDescription $packet) {
		$!decoder = ResultSet::Decoder.new($!client.typemap, $packet.fields, True);
	}
	multi method incoming-message(Packet::NoData $) {
	}
	method finished() {
		$!result.keep(PreparedStatement.new(:$!name, :$!client, :@!input-types, :$!decoder)) unless $!result;
	}
	method !failed-description(--> 'Could not prepare') {}
}

my class Protocol::Execute does Protocol::ExtendedQuery {
	multi method incoming-message(Packet::BindComplete $) {
		$!stage = Executing;
		if $!source {
			$!result.keep($!source.resultset);
		}
	}
}
my class Protocol::Close does Protocol {
	has Promise $.result is required handles(:failed-exception<break>);
	multi method incoming-message(Packet::CloseComplete $) {
	}
	method finished() {
		$!result.keep unless $!result;
	}
	method !failed-description(--> 'Could not close prepared statement') {}
}

class Notification {
	has Int:D $.sender is required;
	has Str:D $.channel is required;
	has Str:D $.message is required handles<Str>;
}

our sub default-typemap() {
	TypeMap::JSON.new;
}

class Client {
	has Protocol $!protocol;
	has QueryStatus $.query-status is built(False) = QueryStatus::Idle;
	has Promise $!startup-promise = Promise.new;
	has Supplier $!outbound-messages handles(:outbound-messages<Supply>) = Supplier.new;
	has Supply $!outbound-data;
	has Supplier %!notification-channel;
	has Packet::Decoder $!decoder = Packet::Decoder.new;
	has Int $!prepare-counter = 0;
	has Promise:D $.disconnected is built(False) = Promise.new;
	has TypeMap $.typemap = default-typemap();
	submethod TWEAK() {
		$!disconnected.then: {
			my $message = $!disconnected ~~ Broken ?? ~$!disconnected.cause !! 'Disconnected';
			if $!protocol {
				$!protocol.failed-client($message);
			}
			for @!tasks -> $ (:$protocol, :@packets) {
				$protocol.failed-client($message);
			}
			for %!notification-channel.values -> $channel {
				$channel.done;
			}
		}
	}

	has Int $!process-id handles(:process-id<self>);
	has Int $!secret-key;
	has Str %!parameters handles(:get-parameter<AT-KEY>);

	my class Task {
		has Protocol:D $.protocol is required;
		has Packet::Base:D @.packets is required;
	}
	has Task @!tasks;

	method !send(@packets) {
		$!outbound-messages.emit(@packets);
	}

	method !send-next($protocol, @packets) {
		$!protocol = $protocol;
		self!send(@packets);
	}

	method !handle-next() {
		with @!tasks.shift -> (:$protocol, :@packets) {
			self!send-next($protocol, @packets);
		} else {
			$!protocol = Nil;
		}
	}

	method !submit(Protocol:D $protocol, @packets) {
		if $!disconnected {
			$protocol.failed-client('Disconnected');
		} elsif $!protocol or not $!startup-promise {
			@!tasks.push(Task.new(:$protocol, :@packets));
		} else {
			self!send-next($protocol, @packets);
		}
	}

	method outbound-data() {
		$!outbound-data //= $!outbound-messages.Supply.map({ [~] @^packets.map(*.encode) });
	}
	method incoming-data(Blob:D $data --> Nil) {
		$!decoder.add-data($data);
		while $!decoder.read-packet -> $packet {
			self.incoming-message($packet);
		}
	}

	multi method incoming-message(Packet::AuthenticationOk $) {
	}
	multi method incoming-message(Packet::NegotiateProtocolVersion $version where $version.unknown-options) {
		$!startup-promise.break(X::Client.new('Unknown options ' ~ $version.unknown-options.join(', ')));
	}
	multi method incoming-message(Packet::NegotiateProtocolVersion $version) {
		$!startup-promise.break(X::Client.new('Unsupported protocol version ' ~ $version.newest-minor-version));
	}
	multi method incoming-message(Packet::BackendKeyData $data) {
		$!process-id = $data.process-id;
		$!secret-key = $data.secret-key;
	}
	multi method incoming-message(Packet::ParameterStatus $status) {
		%!parameters{$status.name} = $status.value;
	}
	multi method incoming-message(Packet::NotificationResponse $packet) {
		with %!notification-channel{$packet.channel} -> $supplier {
			$supplier.emit(Notification.new(:sender($packet.sender), :channel($packet.channel), :message($packet.message)));
		}
	}
	multi method incoming-message(Packet::ErrorResponse $error) {
		$!protocol.failed-server($error.values) with $!protocol;
	}
	multi method incoming-message(Packet::ReadyForQuery $ready) {
		$!protocol.finished;
		$!query-status = $ready.status;
		self!handle-next;
	}
	multi method incoming-message(Packet::Base $packet) {
		$!protocol.incoming-message($packet) with $!protocol;
	}

	method get-channel(Str $name --> Supply) {
		my $supplier = %!notification-channel{$name} //= Supplier::Preserving.new;
		$supplier.Supply;
	}

	method startTls(--> Blob) {
		OpenPacket::SSLRequest.new.encode;
	}

	method startup(Str:D $user!, Str $database, Str $password --> Promise) {
		die X::Client.new('Already started') if $!startup-promise or $!protocol;
		my $authenticator = $password.defined ?? Authenticator::Password.new(:$user, :$password) !! Authenticator::Null.new;
		my %parameters = :$user, :DateStyle<ISO>, :client_encoding<UTF8>;
		%parameters<database> = $database with $database;
		my &send-messages = { self!send(@^messages) };
		$!protocol = Protocol::Authenticating.new(:client(self), :$authenticator, :$!startup-promise, :&send-messages);
		self!send([ OpenPacket::StartupMessage.new(:%parameters) ]);
		$!startup-promise;
	}

	method !add-static-type(Type $type, Any:U :$type-object = $type.type-object, Int:D :$oid = $type.oid --> Nil) {
		$!typemap does TypeMap::Overlay unless $!typemap ~~ TypeMap::Overlay;
		$!typemap.add-type($type, :$type-object, :$oid);
	}

	method !add-dynamic-type(Str $name, &callback --> Promise) {
		self.query('SELECT oid, typarray FROM pg_type WHERE typname = $1', [ $name.lc ]).then: -> $p {
			my $result = await $p;
			my ($oid, $array-oid) = $result.array or die X::Client.new("No such type '$name'");
			my $type = callback($oid);
			self!add-static-type(Type::Array[$type, $array-oid]) with $array-oid;
			self!add-static-type($type, :$oid);
		}
	}
	method add-enum-type(Str $name, ::Enum --> Promise) {
		self!add-dynamic-type($name, -> $oid { Type::Enum[Enum, $oid] });
	}
	method add-composite-type(Str $name, ::Composite, Bool :$positional = constructor-is-positional(Composite) --> Promise) {
		my Pair @attributes = Composite.^attributes.map: { $^attr.name.subst(/ ^ <[$@%&]> '!'? /, '') => $!typemap.for-type($^attr.type) };
		self!add-dynamic-type($name, -> $oid { Type::Composite[Composite, $oid, @attributes, $positional] });
	}
	method add-custom-type(Str $name, ::Custom, &from-string = -> Custom(Str) $custom { $custom }, &to-string = -> Custom $custom { ~$custom }) {
		self!add-dynamic-type($name, -> $oid { Type::Custom[Custom, $oid, &from-string, &to-string] });
	}

	method query-multiple(Str $query --> Supply) {
		my $supplier = Supplier::Preserving.new;
		self!submit(Protocol::Query.new(:$supplier), [ Packet::Query.new(:$query) ]);
		$supplier.Supply;
	}

	sub compress-oids(@oids) {
		all(@oids) == 0 ?? () !! @oids;
	}

	method query(Str:D $query, @values? --> Promise) {
		my $result = Promise.new;
		my &send-messages = { self!send(@^messages) };
		my $protocol = Protocol::BindingQuery.new(:$result, :$!typemap, :&send-messages);

		my @types = $!typemap.for-types(@values);
		my @oids = compress-oids(@types».oid);
		my @formats = compress-formats(@types».format);
		my @fields = @types Z[&type-encode] @values;

		self!submit($protocol, [
			Packet::Parse.new(:$query, :@oids), Packet::Bind.new(:@formats, :@fields),
			Packet::Describe.new, Packet::Execute.new, Packet::Sync.new,
		]);
		$result;
	}

	method prepare(Str:D $query, Str:D :$name = "prepared-{++$!prepare-counter}", :@input-types --> Promise) {
		my $result = Promise.new;
		my @types = $!typemap.for-types(@input-types);
		my @oids = compress-oids(@types».oid);
		my $protocol = Protocol::Prepare.new(:client(self), :$name, :$result);
		self!submit($protocol, [
			Packet::Parse.new(:$query, :$name, :@oids), Packet::Describe.new(:$name, :type(Prepared)), Packet::Sync.new,
		]);
		$result;
	}

	method execute-prepared(Packet::Bind:D $bind, ResultSet::Decoder $decoder --> Promise) {
		my $result = Promise.new;
		my &send-messages = { self!send(@^messages) };
		my $source = $decoder ?? $decoder.make-source !! ResultSet::Source;
		my $protocol = Protocol::Execute.new(:$result, :$source, :&send-messages);

		self!submit($protocol, [ $bind, Packet::Execute.new, Packet::Sync.new ]);
		$result;
	}

	method close-prepared(Str:D $name) {
		my $result = Promise.new;
		my $protocol = Protocol::Close.new(:$result);
		self!submit($protocol, [ Packet::Close.new(:$name), Packet::Sync.new ]);
		$result;
	}

	method terminate(--> Nil) {
		self!send([ Packet::Terminate.new ]);
		$!outbound-messages.done;
	}
}

=begin pod

=head1 Name

Protocol::Postgres - a sans-io postgresql client

=head1 Synopsis

=begin code :lang<raku>

use v6.d;
use Protocol::Postgres;

my $socket = await IO::Socket::Async.connect($host, $port);
my $client = Protocol::Postgres::Client.new;
$socket.Supply(:bin).act({ $client.incoming-data($^data) });
$client.outbound-data.act({ $socket.write($^data) });

await $client.startup($user, $database, $password);

my $resultset = await $client.query('SELECT * FROM foo WHERE id = $1', [ 42 ]);
react {
	whenever $resultset.hash-rows -> (:$name, :$description, :$id) {
		say "$name is $description";
	}
}

=end code

=head1 Description

Protocol::Postgres is sans-io implementation of (the client side of) the postgresql protocol. It is typically used through the C<Protocol::Postgres::Client> class.

=head1 Client

C<Protocol::Postgres::Client> has the following methods

=head2 new(--> Protocol::Postgres::Client)

This creates a new postgres client. It supports one optional named argument:

=begin item1
TypeMap :$typemap = TypeMap::JSON

This is the typemap that is used to translate between Raku's and Postgres' typesystem. The default mapping supports common built-in types such as strings, numbers, bools, dates, datetimes, blobs, arrays and hashes. Other options include C<TypeMap::Native> if you want arrays to map to postgres' native arrays and C<TypeMap::Minimal> if one wants all values to map to strings.

=end item1

=head2 outgoing-data(--> Supply)

This returns a C<Supply> of C<Blob>s to be written to the server.

=head2 incoming-data(Blob --> Nil)

This consumes bytes received from the server.

=head2 startup($user, $database?, $password? --> Promise)

This starts the handshake to the server. C<$database> may be left undefined, the server will use C<$user> as database name. If a C<$password> is defined, any of clearnext, md5 or SCRAM-SHA-256 based authentication is supported.

The resulting promise will finish when the connection is ready for queries.

=head2 query($query, @bind-values --> Promise)

This will issue a query with the given bind values, and return a promise to the result.

For fetching queries such as C<SELECT> the result in the promise will be a C<ResultSet> object, for manipulation (e.g. C<INSERT>) and definition (e.g. C<CREATE>) queries it will result a string describing the change (e.g. C<DELETE 3>). For a C<COPY FROM> query it will C<Supply> with the data stream, and for C<COPY TO> it will be a C<Supplier>.

Both the input types and the output types will be typemapped between Raku types and Postgres types using the typemapper.

=head2 query-multiple($query --> Supply[ResultSet])

This will issue a complex query that may contain multiple statements, but can not use bind values. It will return a C<Supply> to the results of each query.

=head2 prepare($query, :@input-types --> Promise[PreparedStatement])

This prepares the query, and returns a Promise to the PreparedStatement object. C<@input-types> can be used to pass on hints about the types you're passing in during C<execute>.

=head2 method get-channel(Str $name --> Supply)

This returns the C<Supply> for the given channel.

=head2 add-enum-type(Str $name, ::Enum --> Promise)

This looks up the C<oid> of postgres enum C<$name>, and adds an appriopriate C<Type> object to the typemap to convert it from/to C<Enum>.

=head2 add-composite-type(Str $name, ::Composite, Bool :$positional --> Promise)

This looks up the C<oid> of the postgres composite type <$name>, and maps it to C<Composite>; if C<$positional> is set it will use positional constructor arguments, otherwise named ones are used; it will use a heuristic by default.

=head2 add-custom-type(Str $name, ::Custom, &from-string?, &to-string?)

This adds a custom converter from postgres type C<$name> from/to Raku type C<Custom>. By default C<&from-string> will do a coercion, and C<&to-string> will do stringification.

=head2 startTls(--> Blob)

This will return the marker that should be written to the server to start upgrading the connection to use TLS. If the server responds with a single C<S> byte the proposal is accepted and the client is expected to initiate the TLS handshake. If the server responds with an C<N> it is rejected, and the connection proceeds in cleartext.

=head2 terminate(--> Nil)

This sends a message to the server to terminate the connection

=head2 disconnected(--> Promise)

This returns a C<Promise> that must be be kept or broken to signal the connection is lost.

=head2 query-status(--> Protocol::Postgres::QueryStatus)

This returns the query status as of the last finished query as a C<