electric

type Config struct {
	// URL is the sync service, for example http://electric:3000.
	URL string

	// Client makes the upstream request.
	//
	// It must not have a Timeout: a live request is a long poll that
	// deliberately hangs until something changes, and a client timeout would
	// cut every subscription at the same interval. Cancellation comes from the
	// request's context instead, so a client that goes away takes its upstream
	// request with it.
	//
	// This is not the only clock a poll runs under, and it is the only one this
	// field decides. The server's own WriteTimeout would cut the answer on the
	// way out however patient this client was; [Proxy.Serve] replaces that one
	// per request with [PollDeadline].
	Client *http.Client

	// Extra are query parameters added to every upstream request, for the
	// credentials a hosted sync service needs.
	Extra url.Values

	// Headers are added to every upstream request, for the same reason.
	Headers http.Header

	// Secret authorises this proxy to the sync service.
	//
	// It is why the sync service can stay off the load balancer. Electric reads it from
	// the query string, so whatever holds it has to be something a browser is not —
	// which is this proxy, and is the arrangement Electric's own documentation asks
	// for. A client's own credential is separate and is never forwarded; see
	// [Proxy.Serve].
	//
	// Empty leaves it off the request entirely rather than sending an empty one, which
	// a secured sync service reads as a wrong one. That is the service `rig db up`
	// starts, which runs without a secret at all.
	//
	// A field of its own rather than an entry in [Config.Extra], and the reason is not
	// tidiness: this package has to know the value in order to take it back out of what
	// it reports. The credential is on the URL of every upstream request, and a
	// transport failure is an *url.Error, which renders that URL — so without this
	// field the secret reaches [Config.OnError] and then a log. Naming it here is what
	// lets every error this proxy reports have it removed.
	//
	// Setting both this and an Extra["secret"] is refused by [New]: two answers to one
	// question, one of which would be silently dropped.
	//
	// A secret stated only in Extra — which is where one had to go before this field
	// existed — is sent from there and redacted all the same. The redaction is not
	// what this field is for, and a project doing that is the one already logging its
	// credential, so it would be the wrong place to hold the line.
	Secret string

	// DB answers a shape when the sync service cannot be reached, by running the
	// shape's own filter against the database the sync service reads.
	//
	// Setting it gives every shape a fallback. Nil is the older behaviour: a
	// sync outage is a 502 and a subscriber with no rows.
	//
	// A [Shape] is a SELECT — a table, a projection, and a parameterized filter
	// — so there is nothing to write per shape and nothing to keep in step. That
	// is the point. A hand-written fallback is a read this package cannot see
	// inside, so a shape scoped to less than its table with a fallback that is
	// not shows a subscriber rows the subscription would have withheld; the read
	// built from [Shape.Where] is the same predicate the sync service was sent,
	// so it cannot diverge from it.
	//
	// Three things it is worth knowing:
	//
	// The cost lands all at once. Every subscriber falls back at the same
	// moment, because what they have in common is the sync service being gone —
	// so an outage is a read per subscriber per shape against the database the
	// sync service was shielding. [Config.MaxSnapshotRows] bounds each one and
	// [Config.SnapshotTimeout] bounds how long it may take.
	//
	// [Where.Raw] now reaches this database. It was always the one place in this
	// package where the caller is responsible for what it writes; until now what
	// it wrote was parsed by the sync service in a grammar of its own, and from
	// here it is concatenated into a statement sent to Postgres.
	//
	// Postgres accepts filters the sync service refuses, which inverts the usual
	// reassurance that a fallback is the narrower of the two. [Where.In] on a
	// column whose type is a Postgres enum is a 400 from the sync service — see
	// [Where.EqText], which exists for that reason — and runs here without
	// complaint. So a scope with that mistake in it fails loudly while the sync
	// service is up and answers with rows while it is down.
	DB DB

	// InitialTimeout bounds the wait on the one request that is not a long poll:
	// a subscriber reading a shape from the beginning.
	//
	// It exists so that a sync service which is running and not answering is
	// treated as unreachable rather than waited on, which is the outage a
	// [Shape.Fallback] is most useful for and the one a transport error does not
	// catch. A live poll's wait is deliberately left unbounded — see Client;
	// what bounds its answer on the way out is [PollDeadline].
	//
	// The wait only: once the answer has started arriving it is copied out
	// whole, up to [PollDeadline]. A large shape read over a slow connection is
	// a transfer rather than an outage, and half of one is worse than either.
	//
	// Required; [DefaultInitialTimeout] is what to write. A negative value is no
	// timeout, which is
	// what a project whose sync service legitimately takes a while to begin
	// answering should set.
	InitialTimeout time.Duration

	// MaxSnapshotRows bounds a fallback snapshot, and refuses past the bound
	// rather than sending part of one.
	//
	// A snapshot is built whole, in memory, per subscriber — and every subscriber
	// falls back at once, because what they have in common is the sync service
	// being gone. Truncating instead would be worse than refusing: a subscriber
	// cannot tell a short answer from a complete one, so a collection quietly
	// missing half its rows would look like a table that had lost them.
	//
	// Required; [DefaultMaxSnapshotRows] is what to write. A negative value is no
	// bound.
	MaxSnapshotRows int

	// SnapshotTimeout bounds one read of [Config.DB].
	//
	// It exists for the reason [Config.InitialTimeout] does, one layer down.
	// Every subscriber falls back at the same moment, so an outage is every one
	// of them queueing on the same pool — and without this, a database that has
	// become slow because of that queue is a request that waits rather than one
	// that gives up and lets the next through.
	//
	// Required; [DefaultSnapshotTimeout] is what to write. A negative value is no
	// timeout, and the
	// request's own context is then the only bound.
	SnapshotTimeout time.Duration

	// BreakerThreshold is how many failures in a row stop this proxy asking.
	//
	// An outage is one process being unreachable, and every request that goes on
	// asking it pays [InitialTimeout] to learn what the request before it already
	// learned — a held goroutine, a held connection, and a subscriber watching a
	// spinner for ten seconds before it is given the snapshot it could have had
	// at once. Past this many consecutive failures the proxy stops asking and
	// answers from here, until [BreakerCooldown] has passed and one request is
	// let through to find out whether the service is back.
	//
	// The trade is a shape whose sync service is fine and whose network is
	// briefly not: it is answered from a fallback, or refused, for as long as one
	// cooldown — where before it would have waited and then usually succeeded. In
	// a row rather than a rate, so a single failure among successes never counts.
	//
	// Required; [DefaultBreakerThreshold] is what to write. A negative value
	// never opens the
	// circuit, which is every request asking however long the outage lasts.
	BreakerThreshold int

	// BreakerCooldown is how long the proxy goes on not asking before it lets one
	// request through to test the sync service.
	//
	// One request, not all of them: the rest are answered from here while it
	// finds out. Nothing polls in the background, so a service that comes back is
	// noticed by the next subscriber through the door rather than a moment
	// earlier by a goroutine of this package's own.
	//
	// Required; [DefaultBreakerCooldown] is what to write. There is no negative
	// answer here to carry, so zero and below are refused alike.
	BreakerCooldown time.Duration

	// OnError is told about a failure that was answered rather than returned:
	// the sync service being unreachable, and a Fallback that refused. And one
	// that could not be answered at all, because the answer had already begun —
	// a response cut short on its way out.
	//
	// It exists because those are the failures a shape endpoint hides. The
	// proxy writes a status and a short line, and without this the cause reaches
	// nobody — there is no logger in this package, for the reason there is none
	// anywhere else in runtime.
	//
	// A request the circuit refused to make is not one of them: it attempted
	// nothing, so it learned nothing worth a second line. What is worth a line is
	// the circuit opening, and that is [OnSyncState].
	//
	// Every error names the [Shape.Table] it came from. This takes a context and
	// an error and nothing else, so a handler cannot add the route afterwards —
	// and without it an application serving four shapes gets four identical lines
	// during one outage.
	OnError func(context.Context, error)

	// OnSyncState is told when the answer to whether the sync service is there
	// changes: false when the circuit opens, true when a request through it
	// succeeds.
	//
	// Twice per outage rather than once per request, which is what makes it the
	// thing to alert on. [Proxy.SyncReachable] is the same answer for a health
	// endpoint or a banner, asked rather than pushed.
	OnSyncState func(ctx context.Context, reachable bool)
}

type DB = dbx.Conn

type Fallback func(ctx context.Context) (Snapshot, error)

type Proxy struct {
	// contains filtered or unexported fields
}

type Row struct {
	// Key identifies the row within the shape. [RowKey] builds one.
	Key string

	// Value is the row: the shape's columns, each in the text form Postgres
	// prints, or nil for NULL. A column the shape does not carry does not belong
	// here — the projection is the same promise on this path as on the other one.
	Value map[string]any
}

type Shape struct {
	// Table is the table to stream, optionally schema-qualified.
	Table string
	// Where is the filter, with $1-style placeholders bound by Params.
	Where string
	// Params are the placeholder values.
	Params []string
	// Columns narrows the projection. Empty streams every column.
	//
	// It is worth setting: a shape carries every column it names to every
	// subscriber, forever, and a password hash in a live-sync stream is a
	// password hash on somebody's laptop.
	Columns []string

	// Key are the table's primary key columns, in the order the table declares
	// them.
	//
	// They identify a row within the shape — see [RowKey] — and are what
	// [Config.DB] needs to answer this shape without being told anything else
	// about it. A key column outside Columns is still read, and still left out
	// of the row: the projection is the promise on that path too.
	//
	// Required for that read, and refused rather than defaulted where it is
	// missing: a snapshot whose rows are all named after the table is one row as
	// far as a subscriber is concerned. A generated shape always carries it.
	Key []string

	// Schema is the electric-schema document for Columns: a JSON object mapping
	// each column to its Postgres type. See [Snapshot.Schema], which is where it
	// ends up.
	//
	// It describes Columns, so a shape that narrows the projection and does not
	// narrow this describes a different set than it sends.
	Schema string

	// NoFallback answers a sync outage with a 502 even where this proxy could
	// have answered from the database.
	//
	// It is for a shape whose value is its freshness rather than its rows. A
	// snapshot of who is looking at this page, that then stops updating, is
	// worth less than an empty list — so presence sets this, and it is a
	// decision rather than an omission.
	NoFallback bool

	// Fallback answers this shape from somewhere other than the database this
	// package would otherwise read.
	//
	// It is the escape hatch, and it wins: a default that overrode it would not
	// be one. Setting it is for a shape that is not a single table's rows — one
	// answered from a materialised view, a cache, or a second database.
	//
	// Whatever narrows Where has to narrow this too. Where is a filter this
	// package sends and can therefore promise; a Fallback is a read it cannot
	// see inside, so a shape scoped to less than its table with a fallback that
	// is not is a subscriber shown rows the subscription would have withheld.
	// The read built from Where has no such gap, which is why it is the default.
	Fallback Fallback
	// contains filtered or unexported fields
}

type Snapshot struct {
	// Rows are the rows the shape's filter admits. Order does not matter.
	Rows []Row

	// Schema is the electric-schema header: a JSON object mapping each column to
	// its Postgres type, which is how a client picks a parser per column.
	//
	// Sending it is not optional in practice. A subscriber that does not get it
	// leaves every value as the string it arrived as, so a timestamp stays text
	// and a count stays text — decoded differently from the same column read
	// over the API, which is the one thing a generated client exists to prevent.
	Schema string
}

type Where struct {
	// contains filtered or unexported fields
}