From 6e58556858b8bbb28e422f85afea79f4d9d5ddcd Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 13 Aug 2026 08:47:15 +0000 Subject: [PATCH] wip(task14): cycle A — migration 018 consumer state (six tables) + outbound/federation/cursor/DLQ stores Co-Authored-By: Claude Fable 5 --- internal/consume/connector.go | 140 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/consume.go | 95 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/dispatch.go | 137 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/jetstream_fake_test.go | 137 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/outer_acceptance_test.go | 453 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/redrive.go | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/state_store.go | 216 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/store_test.go | 295 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/db/migrations/018_consumer_state.sql | 185 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/federation_prefs.go | 87 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/interfaces.go | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/migrations_test.go | 44 +++++++++++++++++++++++++++++++++++++++++++- internal/store/models.go | 121 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/outbound_objects.go | 152 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/outbound_test.go | 510 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/store/outbound_votes.go | 161 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 16 file(s) changed, 2888 insertion(s)(+), 1 deletion(s)(-) diff --git a/internal/consume/connector.go b/internal/consume/connector.go new file mode 100644 --- /dev/null +++ b/internal/consume/connector.go @@ -0,0 +1,140 @@ +package consume + +import ( + "context" + "sync" + "sync/atomic" + "time" +) + +// STUB (task 14 RED): the shape is the Coves Connector's, the behavior is +// not implemented yet. Port connector.go's dial/reconnect/cursor loop, the +// in-line retry schedule, and the dead-letter-write-failure-blocks-cursor +// rule verbatim. + +// CursorStore persists the per-consumer Jetstream cursor (time_us of the last +// fully processed event) so the consumer resumes where it left off instead of +// at the live tail. The schema version is baked into the store, not passed +// per call: the connector has no business knowing about handler versioning. +type CursorStore interface { + // GetCursor returns the persisted cursor for the consumer, or 0 when the + // consumer has never persisted one (first run → live tail). + GetCursor(ctx context.Context, consumerName string) (int64, error) + // SaveCursor upserts the cursor for the consumer. The write is monotonic: + // a smaller value than the stored one is a no-op, never a rewind. + SaveCursor(ctx context.Context, consumerName string, cursorTimeUS int64) error +} + +// DeadLetterWriter captures events that failed all in-line retries so they can +// be replayed later instead of being silently dropped. +type DeadLetterWriter interface { + // AddDeadLetter stores the raw event bytes. redriveAttempts seeds the + // redrive budget: 0 for transient failures, MaxRedriveAttempts for + // permanent ones. Re-adding an already-captured event (same consumer, + // time_us and payload) must succeed as a no-op so the cursor can advance. + AddDeadLetter(ctx context.Context, consumerName string, eventTimeUS int64, eventData []byte, handleErr string, redriveAttempts int) error +} + +// ConnectorStatus is a point-in-time snapshot of a connector's health, fed to +// /admin/metrics (consumer lag, last-event age, reconnects, DLQ depth). +type ConnectorStatus struct { + Name string `json:"name"` + Connected bool `json:"connected"` + ConnectedSince *time.Time `json:"connectedSince,omitempty"` + DisconnectedSince *time.Time `json:"disconnectedSince,omitempty"` + LastEventAt *time.Time `json:"lastEventAt,omitempty"` + CursorTimeUS int64 `json:"cursorTimeUs"` + PersistedCursorTimeUS int64 `json:"persistedCursorTimeUs"` + EventsProcessed uint64 `json:"eventsProcessed"` + EventsDeadLettered uint64 `json:"eventsDeadLettered"` + Reconnects uint64 `json:"reconnects"` + // LastError is excluded from JSON on purpose: raw error strings leak + // hosts, SQL fragments and file paths. + LastError string `json:"-"` + LastErrorAt *time.Time `json:"-"` +} + +// Connector maintains the WebSocket connection to the self-hosted Jetstream +// and feeds events to an EventHandler. +type Connector struct { + name string + wsURL string + handler EventHandler + cursorStore CursorStore + deadLetters DeadLetterWriter + + reconnectDelay time.Duration + cursorFlushInterval time.Duration + cursorRewind time.Duration + retryDelays []time.Duration + + started atomic.Bool + + mu sync.Mutex + status ConnectorStatus +} + +// ConnectorOption configures a Connector. +type ConnectorOption func(*Connector) + +// WithCursorStore enables cursor persistence. Without it the connector +// live-tails, which is only ever appropriate in tests. +func WithCursorStore(store CursorStore) ConnectorOption { + return func(c *Connector) { c.cursorStore = store } +} + +// WithDeadLetterWriter enables the dead letter queue for events that fail all +// in-line retries. +func WithDeadLetterWriter(writer DeadLetterWriter) ConnectorOption { + return func(c *Connector) { c.deadLetters = writer } +} + +// WithReconnectDelay overrides the delay between reconnect attempts. +func WithReconnectDelay(d time.Duration) ConnectorOption { + return func(c *Connector) { c.reconnectDelay = d } +} + +// WithCursorFlushInterval overrides how often the in-memory cursor is +// persisted to the CursorStore. +func WithCursorFlushInterval(d time.Duration) ConnectorOption { + return func(c *Connector) { c.cursorFlushInterval = d } +} + +// WithHandlerRetryDelays overrides the in-line retry schedule for handler +// errors. len(delays)+1 total attempts are made. +func WithHandlerRetryDelays(delays []time.Duration) ConnectorOption { + return func(c *Connector) { c.retryDelays = delays } +} + +// NewConnector creates the Jetstream connector for the named consumer. The +// name keys the persisted cursor and dead-letter rows, so it must be stable +// across releases. +func NewConnector(name, wsURL string, handler EventHandler, opts ...ConnectorOption) *Connector { + c := &Connector{ + name: name, + wsURL: wsURL, + handler: handler, + reconnectDelay: 5 * time.Second, + cursorFlushInterval: 5 * time.Second, + cursorRewind: 5 * time.Second, + retryDelays: []time.Duration{200 * time.Millisecond, time.Second, 3 * time.Second}, + status: ConnectorStatus{Name: name}, + } + for _, opt := range opts { + opt(c) + } + return c +} + +// Start runs the connector until ctx is cancelled, reconnecting on errors and +// flushing the cursor on the way out. +func (c *Connector) Start(ctx context.Context) error { + return nil // STUB +} + +// Status returns a snapshot of the connector's health. +func (c *Connector) Status() ConnectorStatus { + c.mu.Lock() + defer c.mu.Unlock() + return c.status // STUB: never advances +} diff --git a/internal/consume/consume.go b/internal/consume/consume.go new file mode 100644 --- /dev/null +++ b/internal/consume/consume.go @@ -0,0 +1,95 @@ +// Package consume is Tidepool's Jetstream consumer (task 14): the atproto +// half of the world the bridge does NOT host. It watches native Coves users' +// repos for federation opt-outs, profile changes, posts, comments and votes, +// and persists the durable OUTBOUND STATE that deletes and Undo are later +// built from (tasks 15-17 consume it through the OutboundEnqueuer seam). +// +// The connector, rev gate, cursor/dead-letter store and redriver are a PORT +// of the Coves AppView's own Jetstream consumer +// (~/Code/coves/internal/atproto/jetstream): same discipline, same failure +// taxonomy, same "a dead-letter write failure blocks cursor advance" rule. +// Where the two diverge it is stated in a comment, not left to be inferred. +package consume + +import ( + "context" + "errors" +) + +// ErrPermanentEvent marks a handler failure as permanent: the event can never +// succeed no matter how often it is retried (validation rejection, a record +// the lexicon forbids). The connector skips both the in-line retries and the +// redrive budget for these — the event is dead-lettered already exhausted and +// kept only for forensics. Unwrapped errors are treated as transient. +var ErrPermanentEvent = errors.New("permanent event failure") + +// ConsumerNative names the single consumer this task ships. It keys the +// persisted rows in consumer_cursors and jetstream_dead_letters, so it MUST +// stay stable across releases: renaming it silently orphans the cursor (the +// consumer restarts at live tail — the exact loss cursors exist to prevent) +// and strands the dead-letter backlog under the old name. +const ConsumerNative = "native" + +// CursorSchemaVersion versions the HANDLER CONTRACT the persisted cursor +// belongs to. consumer_cursors is keyed (consumer_name, schema_version) so a +// future incompatible handler can replay the retained store from scratch +// without overwriting the production cursor; the two rows coexist. +const CursorSchemaVersion = 1 + +// The collections this consumer subscribes to (Jetstream wantedCollections). +const ( + CollectionFederation = "social.coves.bridge.federation" + CollectionProfile = "social.coves.actor.profile" + CollectionPostV2 = "social.coves.community.postv2" + CollectionComment = "social.coves.community.comment" + CollectionVote = "social.coves.feed.vote" +) + +// JetstreamEvent is one frame off the Jetstream WebSocket. +type JetstreamEvent struct { + Account *AccountEvent `json:"account,omitempty"` + Identity *IdentityEvent `json:"identity,omitempty"` + Commit *CommitEvent `json:"commit,omitempty"` + DID string `json:"did"` + Kind string `json:"kind"` + TimeUS int64 `json:"time_us"` +} + +// AccountEvent is a #account status change. Status is parsed and PERSISTED +// (decision 19): "deleted" is the only value that means deletion, every other +// inactive state is transient. +type AccountEvent struct { + DID string `json:"did"` + Time string `json:"time"` + Seq int64 `json:"seq"` + Active bool `json:"active"` + Status string `json:"status,omitempty"` +} + +// IdentityEvent is a #identity handle change. The handle here may be stale; +// the local part is frozen at actor creation regardless. +type IdentityEvent struct { + DID string `json:"did"` + Handle string `json:"handle"` + Time string `json:"time"` + Seq int64 `json:"seq"` +} + +// CommitEvent is a record write in a repo. A DELETE carries DID, collection +// and rkey ONLY — no record body and no CID. That absence is the whole reason +// store.OutboundObjects exists. +type CommitEvent struct { + Rev string `json:"rev"` + Operation string `json:"operation"` // create | update | delete + Collection string `json:"collection"` + RKey string `json:"rkey"` + Record map[string]any `json:"record,omitempty"` + CID string `json:"cid,omitempty"` +} + +// EventHandler processes a single Jetstream event. Handlers MUST be +// idempotent: cursor rewinds on reconnect intentionally replay a few seconds +// of already-processed events, and a full replay must change nothing. +type EventHandler interface { + HandleEvent(ctx context.Context, event *JetstreamEvent) error +} diff --git a/internal/consume/dispatch.go b/internal/consume/dispatch.go new file mode 100644 --- /dev/null +++ b/internal/consume/dispatch.go @@ -0,0 +1,137 @@ +package consume + +import ( + "context" + "database/sql" + "log/slog" + + "tidepool/internal/store" +) + +// STUB (task 14 RED): the seams and the deterministic-id contract are pinned +// here; every handler body is still missing. + +// ActorMinter is the lazy-mint seam onto task 13's personas service. +// *personas.Service satisfies it. Minting happens at the FIRST federating +// interaction (a comment, a vote) — never eagerly on an opt-in event. +type ActorMinter interface { + CreateActorForDID(ctx context.Context, did, handle string) (*store.APActor, error) +} + +// Intent is one outbound activity the consumer decided on. Intents are TYPED +// rather than pre-rendered AP: translation into ActivityPub vocabulary belongs +// to task 15, which owns the wire format. +type Intent interface { + // ActivityID is the deterministic AP activity id this intent will be + // delivered under (see ActivityID). + ActivityID() string +} + +// CommentIntent is a Create/Update/Delete of a native comment. +type CommentIntent struct { + // Op is the commit operation: create, update or delete. + Op string + // ATURI is the comment record's at-uri. + ATURI string + // ID is the deterministic activity id. + ID string + // CommunityAPID is the target community's AP Group id. + CommunityAPID string + // ParentAPID is the AP object id of the thing replied to, resolved + // through ap_objects (either origin). + ParentAPID string + // Snapshot is the translated state a Delete is rebuilt from — the delete + // commit itself carries no record body. + Snapshot []byte +} + +// ActivityID reports the deterministic activity id. +func (i CommentIntent) ActivityID() string { return i.ID } + +// VoteIntent is a Like/Dislike, or the Undo of one. +type VoteIntent struct { + // Op is the commit operation: create or delete. + Op string + // VoteATURI is the vote record's at-uri — the delete path's lookup key. + VoteATURI string + // SubjectAPID is the AP object id of the thing voted on. + SubjectAPID string + // Direction is up or down, read back from outbound_votes on the delete + // path (the delete commit tells us nothing else). + Direction string + // ID is the deterministic activity id. + ID string + // CommunityAPID is the target community's AP Group id. + CommunityAPID string +} + +// ActivityID reports the deterministic activity id. +func (i VoteIntent) ActivityID() string { return i.ID } + +// OutboundEnqueuer is the task 15 seam. main.go wires a logging noop until +// task 15 swaps in the real delivery queue. +type OutboundEnqueuer interface { + // EnqueueActivity hands one intent to delivery. orderingKey serializes + // causally related work; parentATURI carries the causal dependency + // (decision 15) so a reply is never delivered before its parent. + EnqueueActivity(ctx context.Context, actorDID, orderingKey, parentATURI string, intent Intent) error +} + +// Options configures a Dispatcher. +type Options struct { + // DB is the bridge database. + DB *sql.DB + // Actors is the lazy-mint seam (task 13's personas service). + Actors ActorMinter + // Enqueuer is the task 15 outbound seam. + Enqueuer OutboundEnqueuer + // UserOrigin is AP_USER_ORIGIN: the origin every deterministic activity + // id is minted under. + UserOrigin string + // Logger receives the drop reasons nobody sees from a metric. + Logger *slog.Logger +} + +// Dispatcher is the EventHandler this consumer runs: it routes one Jetstream +// event to the right per-collection handler, filters Tidepool-hosted repos, +// and is idempotent under full replay. +type Dispatcher struct { + db *sql.DB + actors ActorMinter + enqueuer OutboundEnqueuer + userOrigin string + logger *slog.Logger +} + +var _ EventHandler = (*Dispatcher)(nil) + +// NewDispatcher builds the event dispatcher. +func NewDispatcher(opts Options) (*Dispatcher, error) { + return &Dispatcher{ + db: opts.DB, + actors: opts.Actors, + enqueuer: opts.Enqueuer, + userOrigin: opts.UserOrigin, + logger: opts.Logger, + }, nil // STUB: no validation +} + +// HandleEvent routes one Jetstream event. +func (d *Dispatcher) HandleEvent(ctx context.Context, event *JetstreamEvent) error { + return nil // STUB +} + +// ActivityID derives the deterministic outbound activity id for one operation +// on one record (decision 12): +// +// {origin}/ap/activity/{sha256(atURI + op + seq)} +// +// It is deterministic on purpose: a redelivery must reuse the id a peer has +// already seen, and a Delete must be buildable long after the record body is +// gone. seq comes from outbound_objects.last_activity_seq / outbound_votes +// .activity_seq — NOT from the CID, because deletes have none. +// +// This is the ONE exported id function; tasks 15/16/17 all call it. +func ActivityID(origin, atURI, op string, seq int) string { + return "" // STUB +} diff --git a/internal/consume/jetstream_fake_test.go b/internal/consume/jetstream_fake_test.go new file mode 100644 --- /dev/null +++ b/internal/consume/jetstream_fake_test.go @@ -0,0 +1,137 @@ +package consume + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/gorilla/websocket" +) + +// fakeJetstream is a minimal stand-in for the self-hosted Jetstream: it +// upgrades the connection, writes a scripted sequence of raw frames, then +// HOLDS THE CONNECTION OPEN and drains reads until the client goes away. +// +// Holding open matters. If the server closed after the script, the connector +// would treat that as a dropped connection and reconnect in a loop, replaying +// the script on every pass — the test would then be measuring reconnect +// behavior instead of handler idempotence. +// +// Frames are RAW BYTES, never marshalled from this package's structs, so the +// tests pin the WIRE shape Jetstream actually emits rather than agreeing with +// our own types about it. +type fakeJetstream struct { + server *httptest.Server + script [][]byte + + mu sync.Mutex + dials int + cursors []string + queries []string +} + +// newFakeJetstream starts a fake Jetstream serving the given frames at +// /subscribe. It is closed when the test finishes. +func newFakeJetstream(t *testing.T, script ...[]byte) *fakeJetstream { + t.Helper() + + fake := &fakeJetstream{script: script} + upgrader := websocket.Upgrader{ + CheckOrigin: func(*http.Request) bool { return true }, + } + + mux := http.NewServeMux() + mux.HandleFunc("/subscribe", func(w http.ResponseWriter, r *http.Request) { + fake.mu.Lock() + fake.dials++ + fake.cursors = append(fake.cursors, r.URL.Query().Get("cursor")) + fake.queries = append(fake.queries, r.URL.RawQuery) + fake.mu.Unlock() + + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("fake jetstream: upgrade: %v", err) + return + } + defer func() { _ = conn.Close() }() + + for _, frame := range fake.script { + if err := conn.WriteMessage(websocket.TextMessage, frame); err != nil { + return // client went away mid-script; nothing to report + } + } + // Hold open. ReadMessage also services the connector's pings, and + // returns as soon as the client closes. + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + }) + + fake.server = httptest.NewServer(mux) + t.Cleanup(fake.server.Close) + return fake +} + +// URL is the ws:// endpoint a Connector dials. +func (f *fakeJetstream) URL() string { + return "ws" + strings.TrimPrefix(f.server.URL, "http") + "/subscribe" +} + +// Dials reports how many times a client connected. +func (f *fakeJetstream) Dials() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.dials +} + +// Cursors returns the cursor query parameter of every dial, in order. +func (f *fakeJetstream) Cursors() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.cursors...) +} + +// recordedIntent is one OutboundEnqueuer call. +type recordedIntent struct { + ActorDID string + OrderingKey string + ParentATURI string + Intent Intent +} + +// recordingEnqueuer is the task 15 seam, recorded. Task 15 owns AP vocabulary; +// this test only cares that the consumer decided on exactly one intent, with +// the right subject and a deterministic id. +type recordingEnqueuer struct { + mu sync.Mutex + calls []recordedIntent +} + +func (e *recordingEnqueuer) EnqueueActivity(_ context.Context, actorDID, orderingKey, parentATURI string, intent Intent) error { + e.mu.Lock() + defer e.mu.Unlock() + e.calls = append(e.calls, recordedIntent{ + ActorDID: actorDID, + OrderingKey: orderingKey, + ParentATURI: parentATURI, + Intent: intent, + }) + return nil +} + +func (e *recordingEnqueuer) Calls() []recordedIntent { + e.mu.Lock() + defer e.mu.Unlock() + return append([]recordedIntent(nil), e.calls...) +} + +func (e *recordingEnqueuer) Len() int { + e.mu.Lock() + defer e.mu.Unlock() + return len(e.calls) +} diff --git a/internal/consume/outer_acceptance_test.go b/internal/consume/outer_acceptance_test.go new file mode 100644 --- /dev/null +++ b/internal/consume/outer_acceptance_test.go @@ -0,0 +1,453 @@ +package consume + +import ( + "context" + "database/sql" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/identity" + "tidepool/internal/personas" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// The world this acceptance test builds. +const ( + acceptUserOrigin = "https://coves.social" + + // A community Tidepool hosts: it has a communities row AND a repo_state + // row, which is what makes its DID "a repo we commit into". + acceptCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + acceptCommunityAPID = "https://lemmy.world/c/technology" + acceptCommunityHost = "lemmy.world" + acceptCommunityName = "technology" + acceptCommunityHead = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + acceptCommunityRev = "3lzhead000001" + acceptRootAuthorDID = "did:plc:ewvi7nxzyoun6zhxrhs64oiz" + acceptRootRKey = "3lzroot2222aa" + acceptRootCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + acceptRootAPID = acceptUserOrigin + "/ap/object/" + acceptRootRKey + acceptRootATURI = "at://" + acceptRootAuthorDID + "/social.coves.community.postv2/" + acceptRootRKey + + // The commenter. This DID has NO row anywhere: no ap_actors, no + // bridged_actors, no communities, no repo_state. Its actor must be minted + // by the act of commenting. + acceptCommenterDID = "did:plc:7iza6de2dwap2sbkpav7c6c6" + acceptCommentRKey = "3lzcmnt3333bb" + acceptCommentCID = "bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4" + acceptCommentRev = "3lzcmntrev001" + acceptCommentATURI = "at://" + acceptCommenterDID + "/social.coves.community.comment/" + acceptCommentRKey + acceptCommentTimeUS = int64(1_775_000_000_000_000) +) + +// acceptKEK seals minted actors' AP RSA keys (32 bytes, AES-256). +var acceptKEK = []byte("0123456789abcdef0123456789abcdef") + +// acceptActivityIDPattern is the deterministic activity id shape (decision +// 12): the user origin, the fixed /ap/activity/ path, and a sha256 digest. +// Deterministic because a redelivery must reuse the id a peer already saw and +// a Delete must be buildable when the record body is long gone. +const acceptActivityIDPattern = `^https://coves\.social/ap/activity/[0-9a-f]{64}$` + +// TestConsumerFederatesAnUnseenNativeComment is the OUTER acceptance test for +// task 14. +// +// GIVEN a bridged community Tidepool hosts, a thread root already mapped in +// ap_objects, and a native commenter whose DID the bridge has never seen, +// WHEN a scripted Jetstream commit creating a social.coves.community.comment +// in that thread arrives over a real WebSocket, +// THEN: +// +// 1. the commenter's AP actor is MINTED lazily (first federating +// interaction — no eager mint anywhere else); +// 2. outbound_objects holds the durable state a later Delete will be built +// from, carrying the community the comment belongs to; +// 3. EXACTLY ONE outbound intent is handed to the task 15 seam, under a +// deterministic activity id; +// 4. the cursor is persisted so a restart resumes instead of live-tailing. +// +// AND THEN, the property everything else rests on: replaying the SAME script +// from cursor 0 into a fresh connector produces ZERO new intents and ZERO +// state changes. That is the rev gate doing its job — stable activity ids +// alone cannot prevent a rewind from resurrecting stale state. +// +// No network: the only endpoint dialled is the httptest fake. +func TestConsumerFederatesAnUnseenNativeComment(t *testing.T) { + conn := consumeTestDB(t) + ctx := context.Background() + + seedBridgedCommunity(t, conn) + seedThreadRoot(t, conn) + requireNoRowsForDID(t, conn, acceptCommenterDID) + + minter := newPersonasService(t, conn) + state := NewPostgresStateStore(conn, CursorSchemaVersion) + objects := store.NewOutboundObjects(conn) + + // ------------------------------------------------------------------- + // Run 1: first sighting. + // ------------------------------------------------------------------- + firstEnqueuer := &recordingEnqueuer{} + runConnector(t, conn, minter, state, firstEnqueuer, acceptCommentTimeUS, + commentCreateFrame(acceptCommentTimeUS, acceptCommentRev)) + + // 1. Lazy mint. The local part's derivation is deliberately NOT asserted + // here: a commit event carries no handle, and where the handle comes + // from is an open design question (see the task report). What the + // bridge MUST NOT do is federate a comment from an identity that + // doesn't exist. + var mintedActors int + require.NoError(t, conn.QueryRowContext(ctx, + `SELECT COUNT(*) FROM ap_actors WHERE did = $1`, acceptCommenterDID).Scan(&mintedActors)) + require.Equal(t, 1, mintedActors, + "the commenter's first federating interaction must lazily mint exactly one AP actor "+ + "for %s (task 13's get-or-create)", acceptCommenterDID) + + // 2. Durable outbound state, keyed by the comment's at-uri. + stored, err := objects.GetByATURI(ctx, acceptCommentATURI) + require.NoError(t, err, + "outbound_objects must hold state for %s: a later delete commit carries no "+ + "record body and no CID, so the Delete can only be built from here", + acceptCommentATURI) + require.NotNil(t, stored) + assert.Equal(t, acceptCommunityDID, stored.CommunityDID, + "the comment's community is resolved through the thread root's ap_objects mapping") + assert.NotEmpty(t, stored.APObjectID, "the AP id the comment federates as must be recorded") + assert.False(t, stored.IsTombstoned(), "a create must not tombstone anything") + + // 3. Exactly one intent, deterministic id. + calls := firstEnqueuer.Calls() + require.Len(t, calls, 1, + "a single comment create must produce exactly one outbound intent, got %d", len(calls)) + call := calls[0] + + assert.Equal(t, acceptCommenterDID, call.ActorDID, "the intent is attributed to the commenter") + assert.Equal(t, acceptRootATURI, call.ParentATURI, + "parentATURI carries the causal dependency (decision 15): the reply must not be "+ + "delivered before the thing it replies to") + assert.NotEmpty(t, call.OrderingKey, "an intent must carry an ordering key") + + intent, ok := call.Intent.(CommentIntent) + require.True(t, ok, "a comment create must enqueue a CommentIntent, got %T", call.Intent) + assert.Equal(t, "create", intent.Op) + assert.Equal(t, acceptCommentATURI, intent.ATURI) + assert.Equal(t, acceptCommunityAPID, intent.CommunityAPID, + "the intent targets the community's AP Group id") + + require.Regexp(t, acceptActivityIDPattern, intent.ActivityID(), + "outbound activity ids are {origin}/ap/activity/{sha256} (decision 12)") + require.Equal(t, ActivityID(acceptUserOrigin, acceptCommentATURI, "create", 0), + intent.ActivityID(), + "the id must come from the ONE exported derivation, seeded with "+ + "last_activity_seq 0 for a create") + + // 4. The cursor survived the run. + persisted := readCursor(t, conn, ConsumerNative, CursorSchemaVersion) + require.GreaterOrEqual(t, persisted, acceptCommentTimeUS, + "the cursor must be persisted past the processed event: a restart that live-tails "+ + "is the exact data loss cursors exist to prevent") + + // Nothing failed quietly. + assert.Zero(t, countRows(t, conn, "jetstream_dead_letters"), + "a well-formed comment in a bridged community must not dead-letter") + + // The rev gate recorded the applied revision. + var gatedRev string + err = conn.QueryRowContext(ctx, + `SELECT rev FROM jetstream_record_revs WHERE record_uri = $1`, acceptCommentATURI).Scan(&gatedRev) + require.NoError(t, err, + "jetstream_record_revs must hold the applied rev for %s — it is the gate a "+ + "replay is rejected against", acceptCommentATURI) + assert.Equal(t, acceptCommentRev, gatedRev) + + // ------------------------------------------------------------------- + // Run 2: full replay from cursor 0 in a fresh connector. + // ------------------------------------------------------------------- + before := snapshotState(t, conn) + + // Cursor 0 = the worst case the FOLLOWUPS note warns about: a cursor + // behind Jetstream's retention replays the ENTIRE store. + _, err = conn.ExecContext(ctx, `DELETE FROM consumer_cursors`) + require.NoError(t, err) + + replayEnqueuer := &recordingEnqueuer{} + runConnector(t, conn, minter, state, replayEnqueuer, acceptCommentTimeUS, + commentCreateFrame(acceptCommentTimeUS, acceptCommentRev)) + + assert.Empty(t, replayEnqueuer.Calls(), + "a full replay must enqueue NOTHING: the rev gate rejects the already-applied "+ + "revision, so the peer is never asked to process the comment twice") + + after := snapshotState(t, conn) + assert.Equal(t, before, after, + "a full replay must leave every row byte-identical — no reseeded seq, no "+ + "refreshed updated_at, no second actor") + + assert.Zero(t, countRows(t, conn, "jetstream_dead_letters"), + "a replayed event is a skip, not a failure: it must not land in the DLQ") +} + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +// consumeTestDB returns a migrated connection with every table this test +// touches emptied. +func consumeTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "ap_actors", "ap_objects", "communities", "repo_state", + "outbound_objects", "outbound_votes", "federation_prefs", + "consumer_cursors", "jetstream_record_revs", "jetstream_dead_letters") + return database +} + +func newPersonasService(t *testing.T, database *sql.DB) *personas.Service { + t.Helper() + custodian, err := identity.NewCustodian(acceptKEK) + require.NoError(t, err, "build custodian") + svc, err := personas.New(personas.Options{ + DB: database, + Custodian: custodian, + UserOrigin: acceptUserOrigin, + }) + require.NoError(t, err, "build personas service") + return svc +} + +// runConnector wires a fresh Connector + Dispatcher over a fresh fake +// Jetstream, runs it until the connector accounts for the scripted event, then +// shuts it down cleanly (which flushes the cursor). +func runConnector( + t *testing.T, + database *sql.DB, + minter ActorMinter, + state *PostgresStateStore, + enqueuer OutboundEnqueuer, + lastEventTimeUS int64, + script ...[]byte, +) { + t.Helper() + + fake := newFakeJetstream(t, script...) + + dispatcher, err := NewDispatcher(Options{ + DB: database, + Actors: minter, + Enqueuer: enqueuer, + UserOrigin: acceptUserOrigin, + }) + require.NoError(t, err, "build dispatcher") + require.NotNil(t, dispatcher) + + connector := NewConnector(ConsumerNative, fake.URL(), dispatcher, + WithCursorStore(state), + WithDeadLetterWriter(state), + WithCursorFlushInterval(20*time.Millisecond), + WithReconnectDelay(50*time.Millisecond), + ) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- connector.Start(ctx) }() + + // The connector's own accounting is the sentinel: the cursor moves past an + // event once it is fully handled (or safely dead-lettered), so waiting on + // it means every assertion below runs against a settled pipeline. + deadline := time.Now().Add(5 * time.Second) + settled := false + for time.Now().Before(deadline) { + if connector.Status().CursorTimeUS >= lastEventTimeUS { + settled = true + break + } + select { + case err := <-done: + cancel() + t.Fatalf("connector Start returned early (%v) without consuming the scripted "+ + "event; dials=%d", err, fake.Dials()) + default: + } + time.Sleep(10 * time.Millisecond) + } + if !settled { + cancel() + <-done + t.Fatalf("timed out after 5s: the connector never accounted for the scripted event "+ + "(want cursor >= %d, got %d; dials=%d). The consumer must dial the Jetstream "+ + "URL, read frames, dispatch them, and advance its cursor.", + lastEventTimeUS, connector.Status().CursorTimeUS, fake.Dials()) + } + + cancel() + select { + case <-done: // the shutdown path flushes the cursor with a fresh context + case <-time.After(5 * time.Second): + t.Fatal("connector did not shut down within 5s of context cancellation") + } +} + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +// commentCreateFrame is the raw Jetstream frame for a comment create. It is +// written as literal JSON, not marshalled from CommitEvent, so it pins the +// WIRE shape rather than agreeing with our own structs. +func commentCreateFrame(timeUS int64, rev string) []byte { + return []byte(fmt.Sprintf(`{ + "did": %q, + "time_us": %d, + "kind": "commit", + "commit": { + "rev": %q, + "operation": "create", + "collection": "social.coves.community.comment", + "rkey": %q, + "cid": %q, + "record": { + "$type": "social.coves.community.comment", + "reply": { + "root": {"uri": %q, "cid": %q}, + "parent": {"uri": %q, "cid": %q} + }, + "content": "first reply from atproto", + "createdAt": "2026-08-12T10:00:00.000Z" + } + } +}`, + acceptCommenterDID, timeUS, rev, acceptCommentRKey, acceptCommentCID, + acceptRootATURI, acceptRootCID, acceptRootATURI, acceptRootCID)) +} + +// seedBridgedCommunity registers the community AND gives it a repo_state row. +// repo_state is the hosted-repo filter's membership test (a PK lookup, not an +// enumeration): a DID with a row there is a repo Tidepool commits into, whose +// events must never be consumed back in — their outbound is enqueued at write +// time, and consuming them too would double-deliver. +func seedBridgedCommunity(t *testing.T, database *sql.DB) { + t.Helper() + ctx := context.Background() + + _, err := store.NewCommunities(database).UpsertCommunity(ctx, store.Community{ + APGroupID: acceptCommunityAPID, + DID: acceptCommunityDID, + PreferredUsername: acceptCommunityName, + Instance: acceptCommunityHost, + FollowState: store.FollowStateAccepted, + }) + require.NoError(t, err, "seed community") + + _, err = database.ExecContext(ctx, + `INSERT INTO repo_state (did, head_cid, rev) VALUES ($1, $2, $3)`, + acceptCommunityDID, acceptCommunityHead, acceptCommunityRev) + require.NoError(t, err, "seed repo_state for the hosted community repo") +} + +// seedThreadRoot maps the post the comment replies to. This is how the +// consumer learns the comment belongs to a bridged community: it resolves +// reply.root through ap_objects rather than trusting anything in the comment. +func seedThreadRoot(t *testing.T, database *sql.DB) { + t.Helper() + published := time.Date(2026, 8, 11, 9, 0, 0, 0, time.UTC) + _, err := store.NewAPObjects(database).PutMapping(context.Background(), store.APObjectMapping{ + APID: acceptRootAPID, + APType: "Page", + OriginInstance: "coves.social", + Origin: store.OriginBridge, + DID: acceptRootAuthorDID, + AuthorDID: acceptRootAuthorDID, + CommunityDID: acceptCommunityDID, + Collection: "social.coves.community.postv2", + RKey: acceptRootRKey, + CID: acceptRootCID, + PublishedAt: &published, + }) + require.NoError(t, err, "seed thread root mapping") +} + +// requireNoRowsForDID proves the commenter really is unseen, so the mint +// assertion cannot pass on a row some other fixture left behind. +func requireNoRowsForDID(t *testing.T, database *sql.DB, did string) { + t.Helper() + ctx := context.Background() + for _, q := range []struct{ what, query string }{ + {"ap_actors", `SELECT COUNT(*) FROM ap_actors WHERE did = $1`}, + {"bridged_actors", `SELECT COUNT(*) FROM bridged_actors WHERE did = $1`}, + {"communities", `SELECT COUNT(*) FROM communities WHERE did = $1`}, + {"repo_state", `SELECT COUNT(*) FROM repo_state WHERE did = $1`}, + {"ap_objects", `SELECT COUNT(*) FROM ap_objects WHERE did = $1`}, + } { + var n int + require.NoError(t, database.QueryRowContext(ctx, q.query, did).Scan(&n)) + require.Zero(t, n, "%s must hold no row for the unseen commenter %s", q.what, did) + } +} + +// --------------------------------------------------------------------------- +// State snapshots (the replay proof) +// --------------------------------------------------------------------------- + +// replaySnapshot is every mutable fact the replay must leave alone. +type replaySnapshot struct { + ActorCount int + ActorLocalPart string + ActorCreatedAt time.Time + ObjectSeq int + ObjectUpdatedAt time.Time + ObjectTombstone bool + ObjectSnapshot string + GateRev string +} + +func snapshotState(t *testing.T, database *sql.DB) replaySnapshot { + t.Helper() + ctx := context.Background() + + var snap replaySnapshot + require.NoError(t, database.QueryRowContext(ctx, + `SELECT COUNT(*) FROM ap_actors WHERE did = $1`, acceptCommenterDID).Scan(&snap.ActorCount)) + require.NoError(t, database.QueryRowContext(ctx, + `SELECT local_part, created_at FROM ap_actors WHERE did = $1`, acceptCommenterDID). + Scan(&snap.ActorLocalPart, &snap.ActorCreatedAt)) + + var tombstonedAt sql.NullTime + var snapshotJSON sql.NullString + require.NoError(t, database.QueryRowContext(ctx, + `SELECT last_activity_seq, updated_at, tombstoned_at, translated_snapshot::text + FROM outbound_objects WHERE at_uri = $1`, acceptCommentATURI). + Scan(&snap.ObjectSeq, &snap.ObjectUpdatedAt, &tombstonedAt, &snapshotJSON)) + snap.ObjectTombstone = tombstonedAt.Valid + snap.ObjectSnapshot = snapshotJSON.String + + require.NoError(t, database.QueryRowContext(ctx, + `SELECT rev FROM jetstream_record_revs WHERE record_uri = $1`, acceptCommentATURI). + Scan(&snap.GateRev)) + return snap +} + +func readCursor(t *testing.T, database *sql.DB, consumerName string, schemaVersion int) int64 { + t.Helper() + var cursor int64 + err := database.QueryRowContext(context.Background(), + `SELECT cursor_time_us FROM consumer_cursors + WHERE consumer_name = $1 AND schema_version = $2`, + consumerName, schemaVersion).Scan(&cursor) + require.NoError(t, err, + "consumer_cursors must hold a row for (%s, %d) after a run", consumerName, schemaVersion) + return cursor +} + +func countRows(t *testing.T, database *sql.DB, table string) int { + t.Helper() + var n int + require.NoError(t, database.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM `+table).Scan(&n)) + return n +} diff --git a/internal/consume/redrive.go b/internal/consume/redrive.go new file mode 100644 --- /dev/null +++ b/internal/consume/redrive.go @@ -0,0 +1,51 @@ +package consume + +import ( + "context" + "time" +) + +// MaxRedriveAttempts is the redrive budget for a dead letter. Rows at or above +// this many attempts are skipped by the redriver but stay in the table for +// manual inspection and still count toward the backlog. The connector +// dead-letters permanent failures (ErrPermanentEvent) with attempts already at +// this value so the redriver never touches them. +const MaxRedriveAttempts = 10 + +// DeadLetterEvent is a dead-lettered Jetstream event awaiting redrive. +type DeadLetterEvent struct { + ID int64 + ConsumerName string + EventTimeUS int64 + EventData []byte + LastError string + Attempts int + CreatedAt time.Time + UpdatedAt time.Time +} + +// DeadLetterQueue is the full dead letter store used by the redriver. +// Connectors only need the narrower DeadLetterWriter. +type DeadLetterQueue interface { + DeadLetterWriter + + // ListRetryable returns up to limit dead letters for the consumer with + // fewer than maxAttempts redrive attempts, oldest first. + ListRetryable(ctx context.Context, consumerName string, maxAttempts, limit int) ([]DeadLetterEvent, error) + + // DeleteDeadLetter removes a successfully redriven event. + DeleteDeadLetter(ctx context.Context, id int64) error + + // MarkRedriveAttempt increments the attempt counter after a failed + // redrive and records the error. + MarkRedriveAttempt(ctx context.Context, id int64, handleErr string) error + + // RetireDeadLetter marks a dead letter permanently exhausted in ONE step + // (attempts jump straight to MaxRedriveAttempts) so it stops consuming + // redrive passes. The row stays for forensics and stays in the backlog + // count. + RetireDeadLetter(ctx context.Context, id int64, reason string) error + + // CountDeadLetters returns the dead letter backlog per consumer. + CountDeadLetters(ctx context.Context) (map[string]int64, error) +} diff --git a/internal/consume/state_store.go b/internal/consume/state_store.go new file mode 100644 --- /dev/null +++ b/internal/consume/state_store.go @@ -0,0 +1,216 @@ +package consume + +import ( + "context" + "database/sql" + "errors" + "fmt" +) + +// The SQL here is a PORT of the Coves AppView's state_store.go — the monotonic +// SaveCursor WHERE clause and the `ON CONFLICT DO NOTHING` dead-letter dedup +// above all — with the one Tidepool divergence: cursors are keyed +// (consumer_name, schema_version), not consumer_name alone. + +// PostgresStateStore persists Jetstream consumer state: per-consumer cursors +// and the dead letter queue. It lives in this package rather than +// internal/store because the state is private to the consumer pipeline — no +// other domain reads these tables. +type PostgresStateStore struct { + db *sql.DB + // schemaVersion is the handler-contract version this store's cursor rows + // belong to. It is fixed per store, never a per-call argument: mixing + // versions inside one consumer run is exactly what the column exists to + // prevent. + schemaVersion int +} + +// NewPostgresStateStore creates a store for consumer state at the given +// handler schema version. +func NewPostgresStateStore(db *sql.DB, schemaVersion int) *PostgresStateStore { + return &PostgresStateStore{db: db, schemaVersion: schemaVersion} +} + +// Compile-time interface satisfaction checks. +var ( + _ CursorStore = (*PostgresStateStore)(nil) + _ DeadLetterQueue = (*PostgresStateStore)(nil) +) + +// GetCursor returns the persisted cursor for the consumer at this store's +// schema version, or 0 if none exists. +func (s *PostgresStateStore) GetCursor(ctx context.Context, consumerName string) (int64, error) { + var cursorTimeUS int64 + err := s.db.QueryRowContext(ctx, + `SELECT cursor_time_us FROM consumer_cursors + WHERE consumer_name = $1 AND schema_version = $2`, + consumerName, s.schemaVersion, + ).Scan(&cursorTimeUS) + if errors.Is(err, sql.ErrNoRows) { + // A first run is not an error: it is a live tail. Note that a NEW + // schema version legitimately lands here even while the production + // version's row sits at the head of the stream — replaying from + // scratch is the whole reason the version bump happened. + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("get cursor for %s (schema %d): %w", consumerName, s.schemaVersion, err) + } + return cursorTimeUS, nil +} + +// SaveCursor upserts the cursor for the consumer at this store's schema +// version. Monotonic: a smaller value than the stored one is a no-op. +func (s *PostgresStateStore) SaveCursor(ctx context.Context, consumerName string, cursorTimeUS int64) error { + // The WHERE on the DO UPDATE is the monotonicity guard: an out-of-order + // flush (a late goroutine, a reconnect rewind) silently does nothing + // rather than walking the cursor backwards over events already accounted + // for — or un-advancing past a poison frame the DLQ already absorbed. + _, err := s.db.ExecContext(ctx, ` + INSERT INTO consumer_cursors (consumer_name, schema_version, cursor_time_us, updated_at) + VALUES ($1, $2, $3, now()) + ON CONFLICT (consumer_name, schema_version) DO UPDATE + SET cursor_time_us = EXCLUDED.cursor_time_us, updated_at = now() + WHERE consumer_cursors.cursor_time_us < EXCLUDED.cursor_time_us`, + consumerName, s.schemaVersion, cursorTimeUS, + ) + if err != nil { + return fmt.Errorf("save cursor for %s (schema %d): %w", consumerName, s.schemaVersion, err) + } + return nil +} + +// AddDeadLetter stores a failed event for later redrive. Re-adding an +// already-captured event is a no-op success (the dedup index absorbs it) so +// the cursor may advance past a poison frame. +func (s *PostgresStateStore) AddDeadLetter(ctx context.Context, consumerName string, eventTimeUS int64, eventData []byte, handleErr string, redriveAttempts int) error { + // event_data is written as raw bytes so byte-corrupt frames are capturable. + // redriveAttempts seeds the budget: 0 for transient failures, and + // MaxRedriveAttempts for permanent ones, which are kept for forensics only. + _, err := s.db.ExecContext(ctx, ` + INSERT INTO jetstream_dead_letters (consumer_name, event_time_us, event_data, last_error, attempts) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT DO NOTHING`, + consumerName, eventTimeUS, eventData, handleErr, redriveAttempts, + ) + if err != nil { + return fmt.Errorf("add dead letter for %s: %w", consumerName, err) + } + return nil +} + +// ListRetryable returns up to limit dead letters for the consumer that have +// not exhausted their redrive attempts, oldest first. +func (s *PostgresStateStore) ListRetryable(ctx context.Context, consumerName string, maxAttempts, limit int) ([]DeadLetterEvent, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, consumer_name, event_time_us, event_data, last_error, attempts, created_at, updated_at + FROM jetstream_dead_letters + WHERE consumer_name = $1 AND attempts < $2 + ORDER BY id ASC + LIMIT $3`, + consumerName, maxAttempts, limit, + ) + if err != nil { + return nil, fmt.Errorf("list retryable dead letters for %s: %w", consumerName, err) + } + defer func() { + _ = rows.Close() // iteration errors surface via rows.Err() + }() + + var deadLetters []DeadLetterEvent + for rows.Next() { + var deadLetter DeadLetterEvent + if err := rows.Scan( + &deadLetter.ID, + &deadLetter.ConsumerName, + &deadLetter.EventTimeUS, + &deadLetter.EventData, + &deadLetter.LastError, + &deadLetter.Attempts, + &deadLetter.CreatedAt, + &deadLetter.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("scan dead letter: %w", err) + } + deadLetters = append(deadLetters, deadLetter) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate dead letters for %s: %w", consumerName, err) + } + return deadLetters, nil +} + +// DeleteDeadLetter removes a successfully redriven event. +func (s *PostgresStateStore) DeleteDeadLetter(ctx context.Context, id int64) error { + // A missing row is success: the redriver may re-run a pass, and the end + // state it wants — the event gone from the queue — already holds. + if _, err := s.db.ExecContext(ctx, + `DELETE FROM jetstream_dead_letters WHERE id = $1`, id); err != nil { + return fmt.Errorf("delete dead letter %d: %w", id, err) + } + return nil +} + +// MarkRedriveAttempt increments the attempt counter after a failed redrive. +func (s *PostgresStateStore) MarkRedriveAttempt(ctx context.Context, id int64, handleErr string) error { + _, err := s.db.ExecContext(ctx, ` + UPDATE jetstream_dead_letters + SET attempts = attempts + 1, last_error = $2, updated_at = now() + WHERE id = $1`, + id, handleErr, + ) + if err != nil { + return fmt.Errorf("mark redrive attempt on dead letter %d: %w", id, err) + } + return nil +} + +// RetireDeadLetter exhausts a dead letter's redrive budget in one step. +func (s *PostgresStateStore) RetireDeadLetter(ctx context.Context, id int64, reason string) error { + // GREATEST, and a jump straight to MaxRedriveAttempts: an event that can + // never succeed (an unparseable frame, a lexicon rejection) must stop + // costing redrive passes after ONE call, not after the whole budget is + // burnt down one attempt at a time. The row STAYS — retiring is about the + // redriver, not about forgetting. + _, err := s.db.ExecContext(ctx, ` + UPDATE jetstream_dead_letters + SET attempts = GREATEST(attempts, $2), last_error = $3, updated_at = now() + WHERE id = $1`, + id, MaxRedriveAttempts, reason, + ) + if err != nil { + return fmt.Errorf("retire dead letter %d: %w", id, err) + } + return nil +} + +// CountDeadLetters returns the dead letter backlog per consumer. +func (s *PostgresStateStore) CountDeadLetters(ctx context.Context) (map[string]int64, error) { + // Exhausted rows are counted too: the redriver ignores them, but an + // operator watching the backlog must still see them. + rows, err := s.db.QueryContext(ctx, ` + SELECT consumer_name, COUNT(*) + FROM jetstream_dead_letters + GROUP BY consumer_name`, + ) + if err != nil { + return nil, fmt.Errorf("count dead letters: %w", err) + } + defer func() { + _ = rows.Close() // iteration errors surface via rows.Err() + }() + + counts := make(map[string]int64) + for rows.Next() { + var consumerName string + var count int64 + if err := rows.Scan(&consumerName, &count); err != nil { + return nil, fmt.Errorf("scan dead letter count: %w", err) + } + counts[consumerName] = count + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate dead letter counts: %w", err) + } + return counts, nil +} diff --git a/internal/consume/store_test.go b/internal/consume/store_test.go new file mode 100644 --- /dev/null +++ b/internal/consume/store_test.go @@ -0,0 +1,295 @@ +package consume + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/testutil" +) + +// Task 14 cycle A: the consumer's own state — cursors and the dead letter +// queue. These mirror the Coves AppView's shapes (state_store.go) with one +// deliberate divergence: cursors are keyed (consumer_name, schema_version). +// +// Why the version matters, learned the hard way (FOLLOWUPS): a cursor sitting +// between Jetstream's newest stored event and now replays the ENTIRE retained +// store. A future incompatible handler must be able to do that replay without +// stomping the production cursor — so the two rows coexist. + +// consumeStateTestDB returns a migrated connection with the consumer's own +// tables emptied. +func consumeStateTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, "consumer_cursors", "jetstream_dead_letters") + return database +} + +const otherConsumer = "other" + +// --------------------------------------------------------------------------- +// consumer_cursors +// --------------------------------------------------------------------------- + +func TestCursorStore_MissingCursorIsZero(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + cursor, err := store.GetCursor(ctx, ConsumerNative) + require.NoError(t, err, "a first run must not be an error, it must be a live tail") + assert.Equal(t, int64(0), cursor) +} + +func TestCursorStore_SaveAndGetRoundTrip(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.SaveCursor(ctx, ConsumerNative, 1_700_000_000_000_000)) + + cursor, err := store.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(1_700_000_000_000_000), cursor) + + // Consumers are isolated from each other. + other, err := store.GetCursor(ctx, otherConsumer) + require.NoError(t, err) + assert.Equal(t, int64(0), other, "another consumer's cursor must be untouched") + + // The row is keyed by BOTH columns. + var count int + require.NoError(t, database.QueryRowContext(ctx, + `SELECT COUNT(*) FROM consumer_cursors WHERE consumer_name = $1 AND schema_version = $2`, + ConsumerNative, CursorSchemaVersion).Scan(&count), + "consumer_cursors must be keyed (consumer_name, schema_version)") + assert.Equal(t, 1, count) +} + +func TestCursorStore_SaveIsMonotonic(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.SaveCursor(ctx, ConsumerNative, 2_000)) + + // An out-of-order flush (a late goroutine, a reconnect rewind) must never + // walk the cursor backwards: that would replay events the consumer has + // already accounted for, or worse, un-advance past a poison frame. + require.NoError(t, store.SaveCursor(ctx, ConsumerNative, 1_000), + "saving a smaller cursor is a no-op, not an error") + cursor, err := store.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(2_000), cursor, "a smaller cursor must not rewind the stored one") + + require.NoError(t, store.SaveCursor(ctx, ConsumerNative, 3_000)) + cursor, err = store.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(3_000), cursor, "a greater cursor advances") +} + +func TestCursorStore_SchemaVersionsCoexist(t *testing.T) { + database := consumeStateTestDB(t) + ctx := context.Background() + + current := NewPostgresStateStore(database, CursorSchemaVersion) + next := NewPostgresStateStore(database, CursorSchemaVersion+1) + + require.NoError(t, current.SaveCursor(ctx, ConsumerNative, 9_000)) + + // The next handler version starts from scratch — it has never processed + // anything, so it must NOT inherit the production cursor. + cursor, err := next.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(0), cursor, + "a new schema version starts at 0: inheriting the old cursor would skip the "+ + "replay the version bump exists to perform") + + // And its own replay must not stomp the production cursor, even though it + // is far behind. + require.NoError(t, next.SaveCursor(ctx, ConsumerNative, 100)) + + cursor, err = current.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(9_000), cursor, + "the production cursor must survive another version's replay") + + cursor, err = next.GetCursor(ctx, ConsumerNative) + require.NoError(t, err) + assert.Equal(t, int64(100), cursor) + + var rows int + require.NoError(t, database.QueryRowContext(ctx, + `SELECT COUNT(*) FROM consumer_cursors WHERE consumer_name = $1`, ConsumerNative).Scan(&rows)) + assert.Equal(t, 2, rows, "two schema versions means two rows for one consumer name") +} + +// --------------------------------------------------------------------------- +// jetstream_dead_letters +// --------------------------------------------------------------------------- + +func TestDeadLetters_ReAddingTheSameEventIsANoOp(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + payload := []byte(`{"kind":"commit","time_us":42,"commit":{"operation":"create"}}`) + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 42, payload, "boom", 0)) + // A poison event replayed by the reconnect rewind hits the dedup index. + // It MUST succeed as a no-op: an error here would block the cursor + // forever on an event that is already safely captured. + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 42, payload, "boom again", 0), + "re-adding an already-captured event must be a no-op success so the cursor can advance") + + counts, err := store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Equal(t, int64(1), counts[ConsumerNative], "one row, not two") + + // A different payload at the same time_us is a different event. + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 42, + []byte(`{"kind":"commit","time_us":42,"commit":{"operation":"delete"}}`), "boom", 0)) + counts, err = store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Equal(t, int64(2), counts[ConsumerNative], + "the dedup key is (consumer, time_us, payload) — a different payload is a different event") + + // And so is the same payload under a different consumer. + require.NoError(t, store.AddDeadLetter(ctx, otherConsumer, 42, payload, "boom", 0)) + counts, err = store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Equal(t, int64(2), counts[ConsumerNative]) + assert.Equal(t, int64(1), counts[otherConsumer], "the backlog is reported per consumer") +} + +func TestDeadLetters_ListRetryableExcludesExhausted(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + transient := []byte(`{"kind":"commit","time_us":1}`) + permanent := []byte(`{"kind":"commit","time_us":2}`) + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, transient, "pg blip", 0)) + // A permanent failure is captured with its budget already spent: replaying + // a validation rejection can never succeed, so the redriver must never + // pick it up. It stays for forensics. + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 2, permanent, + "lexicon rejection", MaxRedriveAttempts)) + + retryable, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, retryable, 1, + "only the transient failure is retryable; the permanent one is already exhausted") + assert.Equal(t, int64(1), retryable[0].EventTimeUS) + assert.Equal(t, transient, retryable[0].EventData, "the RAW frame is what gets replayed") + assert.Equal(t, "pg blip", retryable[0].LastError) + assert.Equal(t, 0, retryable[0].Attempts) + assert.Equal(t, ConsumerNative, retryable[0].ConsumerName) + assert.NotZero(t, retryable[0].ID) + + // The exhausted row still counts toward the backlog operators watch. + counts, err := store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Equal(t, int64(2), counts[ConsumerNative], + "an exhausted row is skipped by the redriver but stays in the backlog count") + + // Another consumer's backlog is never handed to this one's redriver. + require.NoError(t, store.AddDeadLetter(ctx, otherConsumer, 3, []byte(`{"time_us":3}`), "x", 0)) + retryable, err = store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + assert.Len(t, retryable, 1, "ListRetryable is scoped to one consumer") +} + +func TestDeadLetters_ListRetryableIsOldestFirstAndBounded(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + for i := int64(1); i <= 3; i++ { + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, i, + []byte(`{"time_us":`+string(rune('0'+i))+`}`), "boom", 0)) + } + + all, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, all, 3) + assert.Equal(t, int64(1), all[0].EventTimeUS, "oldest first: events replay in arrival order") + assert.Equal(t, int64(3), all[2].EventTimeUS) + + page, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 2) + require.NoError(t, err) + assert.Len(t, page, 2, "limit bounds the batch") +} + +func TestDeadLetters_MarkRedriveAttempt(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`{"time_us":1}`), "first", 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + require.NoError(t, store.MarkRedriveAttempt(ctx, listed[0].ID, "still failing")) + + listed, err = store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1, "one burnt attempt does not exhaust the budget") + assert.Equal(t, 1, listed[0].Attempts) + assert.Equal(t, "still failing", listed[0].LastError, "the newest error replaces the old one") + + // maxAttempts is the caller's cutoff, evaluated against the counter. + listed, err = store.ListRetryable(ctx, ConsumerNative, 1, 10) + require.NoError(t, err) + assert.Empty(t, listed, "attempts >= maxAttempts is excluded") +} + +func TestDeadLetters_RetireExhaustsInOneStep(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`not json at all`), "parse", 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + // An unparseable payload can never succeed. Retiring it must cost ONE + // call, not MaxRedriveAttempts redrive passes. + require.NoError(t, store.RetireDeadLetter(ctx, listed[0].ID, "unparseable event")) + + retryable, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + assert.Empty(t, retryable, "a retired dead letter is exhausted after exactly one call") + + counts, err := store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Equal(t, int64(1), counts[ConsumerNative], + "the row is KEPT for forensics and stays in the backlog count") +} + +func TestDeadLetters_DeleteRemovesRedrivenEvent(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`{"time_us":1}`), "boom", 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + require.NoError(t, store.DeleteDeadLetter(ctx, listed[0].ID)) + + counts, err := store.CountDeadLetters(ctx) + require.NoError(t, err) + assert.Zero(t, counts[ConsumerNative], "a successfully redriven event leaves the queue") + + require.NoError(t, store.DeleteDeadLetter(ctx, listed[0].ID), + "deleting an already-deleted dead letter is a no-op success") +} diff --git a/internal/db/migrations/018_consumer_state.sql b/internal/db/migrations/018_consumer_state.sql new file mode 100644 --- /dev/null +++ b/internal/db/migrations/018_consumer_state.sql @@ -0,0 +1,185 @@ +-- +goose Up +-- Task 14: the Jetstream consumer's own state, plus the durable OUTBOUND +-- state every later Delete and Undo is rebuilt from. +-- +-- The consumer half (consumer_cursors, jetstream_record_revs, +-- jetstream_dead_letters) is a PORT of the Coves AppView's own Jetstream +-- tables (their migrations 032/033). Same discipline, same failure taxonomy, +-- one deliberate divergence — stated on consumer_cursors below. + +-- Cursor persistence. Each consumer stores the time_us of the last event it +-- fully processed (handled or safely dead-lettered) and resumes there instead +-- of at the live tail, so restarts, deploys and crashes stop losing the events +-- that happened during the gap. +-- +-- DIVERGENCE FROM COVES: the key is (consumer_name, schema_version), not +-- consumer_name alone. A future incompatible handler must be able to replay +-- the whole retained store from scratch, and it cannot do that if its replay +-- cursor overwrites the production one — the two rows coexist and neither +-- version can see the other's progress. Learned the hard way (FOLLOWUPS): a +-- cursor sitting between Jetstream's newest stored event and now replays the +-- ENTIRE retained store, so every path downstream of a cursor must be +-- idempotent and a wall-clock "now" cursor is never a dedupe boundary. +CREATE TABLE consumer_cursors ( + consumer_name TEXT NOT NULL, + schema_version INT NOT NULL, -- the HANDLER contract version, not the DB schema + cursor_time_us BIGINT NOT NULL CHECK (cursor_time_us >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (consumer_name, schema_version) +); + +-- Per-record rev gate: the ordering guard that makes replay safe. +-- +-- Every commit event carries `rev`, the repo's monotonic TID — a fixed-length +-- base32-sortable string, so plain lexicographic comparison IS commit order +-- within one repo. Handlers record the rev of the last APPLIED event per +-- record URI here and apply an incoming create/update/delete only when its rev +-- is strictly greater. Equal rev means the same event replayed (reconnect +-- rewind, redrive) and is a no-op; smaller means a stale copy and is skipped. +-- +-- WHY A SEPARATE TABLE rather than a rev column on outbound_objects: the row +-- must SURVIVE the record it describes, and it must gate record types that +-- have no outbound row at all (federation prefs, profiles, votes). It doubles +-- as the tombstone that rejects the stale create which would otherwise +-- resurrect a deleted record — stable activity ids alone cannot prevent that. +CREATE TABLE jetstream_record_revs ( + record_uri TEXT PRIMARY KEY, + -- COLLATE "C" pins the comparison to bytewise order (for TIDs, bytewise IS + -- commit order) regardless of the database's default collation. + rev TEXT COLLATE "C" NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Dead letter queue for events that failed all in-line retries. Malformed +-- records go HERE, not to a log line: a lexicon rollout mistake must be +-- recoverable, not silent permanent loss. +-- +-- event_data is BYTEA, not TEXT/JSONB, because byte-corrupt frames (NUL bytes, +-- invalid UTF-8) must also be capturable. A TEXT column would reject them, the +-- failed dead-letter write would tear the connection down without advancing +-- the cursor, and the consumer would replay the same frame forever. +CREATE TABLE jetstream_dead_letters ( + id BIGSERIAL PRIMARY KEY, + consumer_name TEXT NOT NULL, + event_time_us BIGINT NOT NULL DEFAULT 0, + event_data BYTEA NOT NULL, + last_error TEXT NOT NULL, + attempts INT NOT NULL DEFAULT 0, -- redrive attempts, not in-line retries + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- The redriver scans per consumer, oldest first, skipping rows that exhausted +-- their budget. +CREATE INDEX idx_jetstream_dead_letters_redrive + ON jetstream_dead_letters (consumer_name, attempts, id); + +-- Dedup. A poison event inside the reconnect rewind window (or an unparseable +-- frame, which never advances the cursor) would otherwise insert a fresh row — +-- each with its own redrive budget — on every reconnect. AddDeadLetter inserts +-- ON CONFLICT DO NOTHING, so an already-captured event counts as success and +-- the cursor may advance past it. The name is the Coves original's: the store +-- and its tests both name it. +CREATE UNIQUE INDEX idx_jetstream_dead_letters_dedup + ON jetstream_dead_letters (consumer_name, event_time_us, md5(event_data)); + +-- outbound_objects is the state a Delete is rebuilt from (decision 14). +-- +-- WHY IT EXISTS: a Jetstream DELETE commit carries the DID, the collection and +-- the rkey and NOTHING else — no record body, no CID (verified against the +-- event model). So every fact a Delete{Note} needs — which AP id it addresses, +-- which community it is addressed to, what the object looked like — must +-- already be at rest here before the delete arrives. +-- +-- Rows are TOMBSTONED, never deleted: the row is what a late replay of the +-- create is rejected against, and task 17 restores content from the snapshot. +-- +-- last_cid/last_rev are PROVENANCE ONLY. The ordering gate is +-- jetstream_record_revs; reading a rev back from here to decide whether to +-- apply an event would be a check→write race by construction. +-- +-- last_activity_seq is the counter behind the deterministic activity id +-- (decision 12: sha256(at_uri + op + seq)). It is a seq and not the CID +-- because deletes have no CID. A create is seq 0; every APPLIED write bumps +-- it, so each operation gets its own stable id — stable because the rev gate +-- runs first and a replayed commit never reaches the bump. +CREATE TABLE outbound_objects ( + at_uri TEXT PRIMARY KEY, + ap_object_id TEXT NOT NULL, -- the AP id this record federates as + last_cid TEXT NOT NULL DEFAULT '', -- provenance only + last_rev TEXT NOT NULL DEFAULT '', -- provenance only + community_did TEXT NOT NULL, + community_ap_id TEXT NOT NULL, + translated_snapshot JSONB NOT NULL, -- what task 15 serves and task 17 restores + last_activity_seq INT NOT NULL DEFAULT 0, + depth INT NOT NULL DEFAULT 0, -- reply depth; Lemmy caps comments at 50 + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + tombstoned_at TIMESTAMPTZ +); + +-- Task 15 serves an actor's outbound work and task 17 sweeps a community's. +CREATE INDEX idx_outbound_objects_community ON outbound_objects (community_did); + +-- outbound_votes is the state an Undo is rebuilt from (decision 16). +-- +-- The PRIMARY KEY is the VOTE RECORD's at-uri because that is the only thing a +-- vote delete commit carries: direction and the activity id the Like/Dislike +-- went out under have to be readable back from that one key alone. +-- +-- (actor_did, subject_at_uri) is a SECOND unique constraint, named EXPLICITLY +-- (postgres would default it to outbound_votes_actor_did_subject_at_uri_key) +-- because the store's 23505 → ConflictError mapping switches on the name. One +-- actor holds at most one LIVE vote per subject: letting a second vote record +-- clobber the first would strand an Undo that is still owed to the peer, so +-- the constraint refuses the write instead. +-- +-- delivered_state is the consumer's INTENT ledger. The consumer only ever +-- writes 'pending'; task 15 flips it on DELIVERY SUCCESS, never on enqueue — +-- a row claiming delivery the wire never confirmed makes the Undo unsendable. +CREATE TABLE outbound_votes ( + vote_at_uri TEXT PRIMARY KEY, + actor_did TEXT NOT NULL, + subject_at_uri TEXT NOT NULL, + subject_ap_id TEXT NOT NULL, + community_did TEXT NOT NULL, + direction TEXT NOT NULL CHECK (direction IN ('up', 'down')), + current_activity_id TEXT NOT NULL, + delivered_state TEXT NOT NULL DEFAULT 'pending' + CHECK (delivered_state IN ('pending', 'delivered', 'undone')), + activity_seq INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT outbound_votes_actor_subject_key UNIQUE (actor_did, subject_at_uri) +); + +-- federation_prefs holds Coves users' federation preferences (decision 11). +-- +-- Federation is DEFAULT-ON and social.coves.bridge.federation is an OPT-OUT +-- record, so ABSENCE IS THE DEFAULT-ON STATE: this table only ever holds rows +-- for users who said something, and the record-delete path removes the row +-- rather than writing enabled = true. Nothing may read a missing row as +-- "unknown" — a caller that cannot tell "opted in" from "never spoke" cannot +-- tell a re-enable from a first sighting either. +-- +-- source is CHECKed and has no default: "we read this from a record" and "we +-- went and asked" have different staleness, and a defaulted source hides which +-- one applied. +CREATE TABLE federation_prefs ( + did TEXT PRIMARY KEY, + enabled BOOL NOT NULL, + delete_remote BOOL NOT NULL DEFAULT FALSE, -- the destructive tier (task 17) + source TEXT NOT NULL CHECK (source IN ('record', 'probe')), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- +goose Down +DROP TABLE IF EXISTS federation_prefs; +DROP TABLE IF EXISTS outbound_votes; +DROP INDEX IF EXISTS idx_outbound_objects_community; +DROP TABLE IF EXISTS outbound_objects; +DROP INDEX IF EXISTS idx_jetstream_dead_letters_dedup; +DROP INDEX IF EXISTS idx_jetstream_dead_letters_redrive; +DROP TABLE IF EXISTS jetstream_dead_letters; +DROP TABLE IF EXISTS jetstream_record_revs; +DROP TABLE IF EXISTS consumer_cursors; diff --git a/internal/store/federation_prefs.go b/internal/store/federation_prefs.go new file mode 100644 --- /dev/null +++ b/internal/store/federation_prefs.go @@ -0,0 +1,87 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "tidepool/internal/errors" +) + +type postgresFederationPrefs struct { + db *sql.DB +} + +// NewFederationPrefs creates the postgres-backed federation_prefs repository. +func NewFederationPrefs(db *sql.DB) FederationPrefs { + return &postgresFederationPrefs{db: db} +} + +const federationPrefColumns = `did, enabled, delete_remote, source, updated_at` + +func (r *postgresFederationPrefs) Upsert(ctx context.Context, pref FederationPref) (*FederationPref, error) { + if !pref.Source.Valid() { + // Including the zero value: an unstated source hides whether the + // preference came off a record we saw or a probe we made, and those + // two have different staleness. + return nil, errors.NewValidationError("source", "unknown source "+string(pref.Source)) + } + + // EVERY field is overwritten, delete_remote included. Re-enabling a user + // must clear a previously requested deleteRemote: a stale destructive flag + // sitting on an enabled row is a loaded gun pointed at task 17. + query := ` + INSERT INTO federation_prefs (did, enabled, delete_remote, source, updated_at) + VALUES ($1, $2, $3, $4, now()) + ON CONFLICT (did) DO UPDATE SET + enabled = EXCLUDED.enabled, + delete_remote = EXCLUDED.delete_remote, + source = EXCLUDED.source, + updated_at = now() + RETURNING ` + federationPrefColumns + + row := r.db.QueryRowContext(ctx, query, + pref.DID, pref.Enabled, pref.DeleteRemote, string(pref.Source)) + stored, err := scanFederationPref(row) + if err != nil { + return nil, fmt.Errorf("upsert federation_pref %q: %w", pref.DID, err) + } + return stored, nil +} + +func (r *postgresFederationPrefs) Get(ctx context.Context, did string) (*FederationPref, error) { + query := `SELECT ` + federationPrefColumns + ` FROM federation_prefs WHERE did = $1` + pref, err := scanFederationPref(r.db.QueryRowContext(ctx, query, did)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + // NotFound MEANS default-on. It is reported as an absence rather + // than synthesized into an enabled row so callers can still tell a + // re-enable from a first sighting. + return nil, errors.NewNotFoundError("federation_pref", did) + } + return nil, fmt.Errorf("get federation_pref %q: %w", did, err) + } + return pref, nil +} + +func (r *postgresFederationPrefs) Delete(ctx context.Context, did string) error { + // Deleting a preference that never existed is success: an opt-out record + // delete for a user who never opted out is the COMMON case, and absence is + // exactly the state the delete is asking for. + if _, err := r.db.ExecContext(ctx, + `DELETE FROM federation_prefs WHERE did = $1`, did); err != nil { + return fmt.Errorf("delete federation_pref %q: %w", did, err) + } + return nil +} + +func scanFederationPref(row rowScanner) (*FederationPref, error) { + var pref FederationPref + var source string + if err := row.Scan(&pref.DID, &pref.Enabled, &pref.DeleteRemote, &source, &pref.UpdatedAt); err != nil { + return nil, err + } + pref.Source = FederationPrefSource(source) + return &pref, nil +} diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -292,6 +292,111 @@ // GetEvent returns the event for an activity id. GetEvent(ctx context.Context, activityID string) (*InboxEvent, error) } +// OutboundObjects persists the state every outbound Delete and Update is +// rebuilt from (task 14, decision 14). A Jetstream delete commit carries the +// DID, collection and rkey and nothing else — no record body, no CID — so a +// Delete{Note} can only be built from what was written here at create time. +// +// Rows are TOMBSTONED, never removed: the row is what a late replay of the +// create is rejected against. +type OutboundObjects interface { + // Upsert idempotently writes the outbound state keyed on ATURI and + // returns the stored row. A new row starts at LastActivitySeq 0; every + // later upsert of the same at-uri bumps it, so each applied operation + // gets its own stable activity id. CreatedAt is preserved. + // + // The bump is safe ONLY because the rev gate runs first: a replayed + // commit never reaches this method, so the seq (and therefore the + // activity id) is stable under replay. + Upsert(ctx context.Context, object OutboundObject) (*OutboundObject, error) + + // UpsertTx is Upsert on an existing transaction — the seam that lets the + // rev-gate claim and the outbound state land in ONE commit. A nil tx is + // an error satisfying errors.IsValidation. + UpsertTx(ctx context.Context, tx *sql.Tx, object OutboundObject) (*OutboundObject, error) + + // GetByATURI returns the outbound state for an at-uri, tombstoned rows + // included (callers check IsTombstoned). A miss is an error satisfying + // errors.IsNotFound. + GetByATURI(ctx context.Context, atURI string) (*OutboundObject, error) + + // Tombstone stamps tombstoned_at, bumps LastActivitySeq and returns the + // full stored row — the state the Delete activity is built from, handed + // back in the same statement that tombstones it so no read/write window + // exists. Tombstoning an already-tombstoned row is a no-op success that + // preserves the original tombstoned_at AND the seq (a redelivered delete + // must reuse the id the first one sent). A missing row is an error + // satisfying errors.IsNotFound. + Tombstone(ctx context.Context, atURI string) (*OutboundObject, error) + + // TombstoneTx is Tombstone on an existing transaction. A nil tx is an + // error satisfying errors.IsValidation. + TombstoneTx(ctx context.Context, tx *sql.Tx, atURI string) (*OutboundObject, error) +} + +// OutboundVotes persists the state an outbound Undo is rebuilt from (decision +// 16). A vote delete commit names the vote record and nothing else, so the +// direction and the activity id the Like/Dislike went out under have to be +// readable back from here. +type OutboundVotes interface { + // Upsert idempotently writes the vote intent keyed on VoteATURI and + // returns the stored row. An empty DeliveredState defaults to + // DeliveredStatePending — the consumer records intent only; delivery is + // task 15's to claim. A new row starts at ActivitySeq 0; re-upserting the + // same vote at-uri bumps it. Writing a DIFFERENT vote at-uri for an + // (ActorDID, SubjectATURI) pair that already has one returns an error + // satisfying errors.IsAlreadyExists: one actor holds at most one live + // vote per subject, and silently clobbering the old row would strand its + // Undo. + Upsert(ctx context.Context, vote OutboundVote) (*OutboundVote, error) + + // UpsertTx is Upsert on an existing transaction. A nil tx is an error + // satisfying errors.IsValidation. + UpsertTx(ctx context.Context, tx *sql.Tx, vote OutboundVote) (*OutboundVote, error) + + // GetByATURI returns the vote for a vote record's at-uri — the DELETE + // path's lookup key, because a delete commit carries nothing else. A miss + // is an error satisfying errors.IsNotFound. + GetByATURI(ctx context.Context, voteATURI string) (*OutboundVote, error) + + // GetByActorSubject returns the actor's live vote on a subject — the + // CREATE path's lookup, which asks "did this actor already vote here?". + // A miss is an error satisfying errors.IsNotFound. + GetByActorSubject(ctx context.Context, actorDID, subjectATURI string) (*OutboundVote, error) + + // SetDeliveredState transitions the delivery state. An unknown state is + // an error satisfying errors.IsValidation; a missing vote is an error + // satisfying errors.IsNotFound. + SetDeliveredState(ctx context.Context, voteATURI string, state DeliveredState) error + + // Delete removes the vote state once its Undo is delivered. Deleting a + // missing vote is a no-op success. + Delete(ctx context.Context, voteATURI string) error +} + +// FederationPrefs stores Coves users' federation preferences (decision 11). +// +// Federation is DEFAULT-ON and social.coves.bridge.federation is an OPT-OUT +// record, so an ABSENT row means enabled. Get therefore returns NotFound for +// a user who never said anything — it never invents an enabled row, because a +// caller that cannot tell "opted in" from "never spoke" cannot tell a +// re-enable from a first sighting either. +type FederationPrefs interface { + // Upsert writes the preference keyed on DID and returns the stored row. + // Every field is overwritten, so re-enabling (Enabled true) also clears a + // previously requested DeleteRemote. Source must be stated explicitly: + // the zero value is an error satisfying errors.IsValidation. + Upsert(ctx context.Context, pref FederationPref) (*FederationPref, error) + + // Get returns the preference for a DID. A miss is an error satisfying + // errors.IsNotFound and MEANS default-on, not "unknown". + Get(ctx context.Context, did string) (*FederationPref, error) + + // Delete removes the preference — the record-delete path, which restores + // the default-on state. Deleting a missing preference is a no-op success. + Delete(ctx context.Context, did string) error +} + // Tombstones remembers AP object ids whose Delete arrived before (or // without) a materialization — the create-after-delete gap: a Create // delivered after its Delete must not resurrect content the origin removed. diff --git a/internal/store/migrations_test.go b/internal/store/migrations_test.go --- a/internal/store/migrations_test.go +++ b/internal/store/migrations_test.go @@ -25,13 +25,35 @@ err := database.QueryRowContext(ctx, ` SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name IN ('ap_objects', 'ap_actors', 'bridged_actors', 'communities', 'inbox_events', 'service_keys', - 'blocks', 'repo_state', 'firehose_events', 'vote_aggregates', 'vote_events') + 'blocks', 'repo_state', 'firehose_events', 'vote_aggregates', 'vote_events', + -- task 14 (migration 018): the consumer's own state and the + -- outbound state deletes are rebuilt from. + 'consumer_cursors', 'jetstream_record_revs', 'jetstream_dead_letters', + 'outbound_objects', 'outbound_votes', 'federation_prefs') `).Scan(&remaining) require.NoError(t, err) assert.Zero(t, remaining, "down migrations must drop every Tidepool table") require.NoError(t, db.MigrateUp(ctx, database), "re-applying up migrations must succeed") + // The down list above only bites if the tables were there to begin with: + // a migration whose Down forgets a table is caught only when its Up + // created one. Assert the newest tables exist after the re-up so the two + // halves stay in step. + for _, table := range []string{ + "consumer_cursors", "jetstream_record_revs", "jetstream_dead_letters", + "outbound_objects", "outbound_votes", "federation_prefs", + } { + var exists bool + err = database.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = $1 + )`, table).Scan(&exists) + require.NoError(t, err) + assert.True(t, exists, "migration 018 must create %q", table) + } + // Leave the schema usable and prove it is: exercise a write. repo := NewAPObjects(database) _, err = repo.PutMapping(ctx, testMapping()) @@ -63,6 +85,26 @@ "communities_did_key", "inbox_events_activity_id_key", "service_keys_name_key", "vote_events_activity_id_key", // the vote dedupe key (task 07) + + // Task 14 (migration 018). The composite cursor key is the whole + // point of consumer_cursors: (consumer_name, schema_version), so a + // future incompatible handler replays without stomping production. + "consumer_cursors_pkey", + "jetstream_record_revs_pkey", + // The dead-letter dedup index — same name as the Coves original this + // is ported from. Without it a poison frame replayed by the reconnect + // rewind grows a fresh row per pass instead of being absorbed, and + // AddDeadLetter stops being the no-op success that lets the cursor + // advance past it. + "idx_jetstream_dead_letters_dedup", + "outbound_objects_pkey", + // outbound_votes is keyed by the VOTE record's at-uri, because that + // is the only thing a vote delete commit carries. The (actor, + // subject) pair is a second, EXPLICITLY NAMED unique constraint: the + // store's 23505 → ConflictError mapping switches on the name. + "outbound_votes_pkey", + "outbound_votes_actor_subject_key", + "federation_prefs_pkey", } for _, name := range expected { var exists bool diff --git a/internal/store/models.go b/internal/store/models.go --- a/internal/store/models.go +++ b/internal/store/models.go @@ -167,6 +167,127 @@ KeyMaterial []byte CreatedAt time.Time } +// DeliveredState tracks how far an outbound vote has travelled. The consumer +// (task 14) only ever writes pending; task 15 flips it on DELIVERY SUCCESS, +// never on enqueue — a state that claimed delivery before the wire confirmed +// it would make an Undo unsendable. +type DeliveredState string + +const ( + // DeliveredStatePending means the intent is recorded but unconfirmed. + DeliveredStatePending DeliveredState = "pending" + // DeliveredStateDelivered means a peer accepted the Like/Dislike. + DeliveredStateDelivered DeliveredState = "delivered" + // DeliveredStateUndone means the Undo was delivered; the row is kept as + // the record of what was withdrawn. + DeliveredStateUndone DeliveredState = "undone" +) + +// Valid reports whether the value is a known delivered state. +func (s DeliveredState) Valid() bool { + switch s { + case DeliveredStatePending, DeliveredStateDelivered, DeliveredStateUndone: + return true + } + return false +} + +// FederationPrefSource records where a federation preference came from: a +// social.coves.bridge.federation record the consumer saw, or a direct probe of +// the user's repo. It is stated explicitly — the zero value is invalid — +// because "we read this from a record" and "we went and asked" have different +// staleness, and a defaulted source hides which one applied. +type FederationPrefSource string + +const ( + // FederationPrefSourceRecord means a Jetstream commit carried the record. + FederationPrefSourceRecord FederationPrefSource = "record" + // FederationPrefSourceProbe means the bridge fetched the record itself. + FederationPrefSourceProbe FederationPrefSource = "probe" +) + +// Valid reports whether the value is a known source. +func (s FederationPrefSource) Valid() bool { + switch s { + case FederationPrefSourceRecord, FederationPrefSourceProbe: + return true + } + return false +} + +// OutboundObject is the durable outbound state for one native record Tidepool +// federates outward (task 14, decision 14). It exists because a Jetstream +// DELETE commit carries the DID, collection and rkey and NOTHING else — no +// record body, no CID — so every fact a Delete{Note} needs must already be at +// rest here before the delete arrives. +type OutboundObject struct { + // ATURI is the record's at-uri and the row's primary key. + ATURI string + // APObjectID is the AP id this record federates as. + APObjectID string + // LastCID and LastRev are PROVENANCE ONLY — what the last applied commit + // looked like. The ordering gate is jetstream_record_revs, never this + // column: a rev read from here is a check→write race by construction. + LastCID string + LastRev string + // CommunityDID and CommunityAPID are the target community on both sides + // of the bridge. + CommunityDID string + CommunityAPID string + // TranslatedSnapshot is the JSONB state task 15 serves the object from and + // task 17 restores it from. + TranslatedSnapshot []byte + // LastActivitySeq feeds ActivityID: create is 0, every applied + // update/delete bumps it, so each operation gets its own stable id. + LastActivitySeq int + // Depth is the reply depth. Lemmy caps comment depth at 50. + Depth int + CreatedAt time.Time + UpdatedAt time.Time + TombstonedAt *time.Time +} + +// IsTombstoned reports whether the record was deleted upstream. Tombstoned +// rows are KEPT: they are what a late replay is rejected against. +func (o *OutboundObject) IsTombstoned() bool { return o.TombstonedAt != nil } + +// OutboundVote is the durable outbound state for one native vote (decision +// 16). A vote DELETE commit names only the vote record, so direction and the +// activity id it was delivered under have to be readable back from here to +// build the Undo. +type OutboundVote struct { + // VoteATURI is the vote record's at-uri and the row's primary key — the + // delete path's only lookup key. + VoteATURI string + // ActorDID and SubjectATURI are UNIQUE TOGETHER: one actor holds at most + // one live vote per subject. + ActorDID string + SubjectATURI string + SubjectAPID string + CommunityDID string + // Direction is up or down. + Direction string + // CurrentActivityID is the id the Like/Dislike went out under; the Undo + // must embed it. + CurrentActivityID string + DeliveredState DeliveredState + // ActivitySeq feeds ActivityID for this vote's operations. + ActivitySeq int + CreatedAt time.Time + UpdatedAt time.Time +} + +// FederationPref is a Coves user's federation preference (decision 11). The +// record is an OPT-OUT and federation is DEFAULT-ON, so an ABSENT row means +// enabled: this table only ever holds rows for users who said something. +type FederationPref struct { + DID string + Enabled bool + DeleteRemote bool + Source FederationPrefSource + UpdatedAt time.Time +} + // InboxEvent is a received AP activity: the dedupe record AND the durable // work-queue item task 06's worker pool consumes. type InboxEvent struct { diff --git a/internal/store/outbound_objects.go b/internal/store/outbound_objects.go new file mode 100644 --- /dev/null +++ b/internal/store/outbound_objects.go @@ -0,0 +1,152 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "tidepool/internal/errors" +) + +type postgresOutboundObjects struct { + db *sql.DB +} + +// NewOutboundObjects creates the postgres-backed outbound_objects repository. +func NewOutboundObjects(db *sql.DB) OutboundObjects { + return &postgresOutboundObjects{db: db} +} + +const outboundObjectColumns = ` + at_uri, ap_object_id, last_cid, last_rev, + community_did, community_ap_id, translated_snapshot, + last_activity_seq, depth, created_at, updated_at, tombstoned_at` + +// execer is the subset of *sql.DB and *sql.Tx these repositories need, so one +// statement runs either standalone or inside a caller's transaction. +type execer interface { + QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +func (r *postgresOutboundObjects) Upsert(ctx context.Context, object OutboundObject) (*OutboundObject, error) { + return r.upsert(ctx, r.db, object) +} + +func (r *postgresOutboundObjects) UpsertTx(ctx context.Context, tx *sql.Tx, object OutboundObject) (*OutboundObject, error) { + if tx == nil { + return nil, errors.NewValidationError("tx", "must not be nil") + } + return r.upsert(ctx, tx, object) +} + +func (r *postgresOutboundObjects) upsert(ctx context.Context, q execer, object OutboundObject) (*OutboundObject, error) { + // created_at is absent from the UPDATE list and last_activity_seq is + // computed from the STORED value rather than the argument: the seq is the + // bridge's own counter for how many activities this object has produced, + // and letting a caller supply it would let a replayed handler reuse — or + // skip — an activity id a peer has already seen. + // + // tombstoned_at is likewise untouched here. A tombstoned row that receives + // a later write keeps its tombstone: un-deleting is a decision for a + // restore path (task 17), not a side effect of an upsert. + query := ` + INSERT INTO outbound_objects ( + at_uri, ap_object_id, last_cid, last_rev, + community_did, community_ap_id, translated_snapshot, depth + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (at_uri) DO UPDATE SET + ap_object_id = EXCLUDED.ap_object_id, + last_cid = EXCLUDED.last_cid, + last_rev = EXCLUDED.last_rev, + community_did = EXCLUDED.community_did, + community_ap_id = EXCLUDED.community_ap_id, + translated_snapshot = EXCLUDED.translated_snapshot, + depth = EXCLUDED.depth, + last_activity_seq = outbound_objects.last_activity_seq + 1, + updated_at = now() + RETURNING` + outboundObjectColumns + + row := q.QueryRowContext(ctx, query, + object.ATURI, object.APObjectID, object.LastCID, object.LastRev, + object.CommunityDID, object.CommunityAPID, object.TranslatedSnapshot, object.Depth, + ) + stored, err := scanOutboundObject(row) + if err != nil { + return nil, fmt.Errorf("upsert outbound_object %q: %w", object.ATURI, err) + } + return stored, nil +} + +func (r *postgresOutboundObjects) GetByATURI(ctx context.Context, atURI string) (*OutboundObject, error) { + query := `SELECT` + outboundObjectColumns + ` FROM outbound_objects WHERE at_uri = $1` + object, err := scanOutboundObject(r.db.QueryRowContext(ctx, query, atURI)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_object", atURI) + } + return nil, fmt.Errorf("get outbound_object %q: %w", atURI, err) + } + return object, nil +} + +func (r *postgresOutboundObjects) Tombstone(ctx context.Context, atURI string) (*OutboundObject, error) { + return r.tombstone(ctx, r.db, atURI) +} + +func (r *postgresOutboundObjects) TombstoneTx(ctx context.Context, tx *sql.Tx, atURI string) (*OutboundObject, error) { + if tx == nil { + return nil, errors.NewValidationError("tx", "must not be nil") + } + return r.tombstone(ctx, tx, atURI) +} + +func (r *postgresOutboundObjects) tombstone(ctx context.Context, q execer, atURI string) (*OutboundObject, error) { + // One statement stamps the tombstone AND returns the state the Delete is + // built from, so no read/write window exists for a concurrent handler to + // interleave in. + // + // Every SET is guarded on tombstoned_at IS NULL, which is what makes a + // redelivered delete a no-op: the seq must NOT bump a second time, because + // the Delete that already went out was addressed with the FIRST seq's + // activity id and the peer must see that same id again. + query := ` + UPDATE outbound_objects SET + tombstoned_at = COALESCE(tombstoned_at, now()), + last_activity_seq = CASE WHEN tombstoned_at IS NULL + THEN last_activity_seq + 1 ELSE last_activity_seq END, + updated_at = CASE WHEN tombstoned_at IS NULL THEN now() ELSE updated_at END + WHERE at_uri = $1 + RETURNING` + outboundObjectColumns + + object, err := scanOutboundObject(q.QueryRowContext(ctx, query, atURI)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + // A delete for a record we never federated. The caller needs to + // tell that apart from a real tombstone: there is no Delete to + // send, and inventing one would address an AP id no peer knows. + return nil, errors.NewNotFoundError("outbound_object", atURI) + } + return nil, fmt.Errorf("tombstone outbound_object %q: %w", atURI, err) + } + return object, nil +} + +func scanOutboundObject(row rowScanner) (*OutboundObject, error) { + var object OutboundObject + var tombstonedAt sql.NullTime + err := row.Scan( + &object.ATURI, &object.APObjectID, &object.LastCID, &object.LastRev, + &object.CommunityDID, &object.CommunityAPID, &object.TranslatedSnapshot, + &object.LastActivitySeq, &object.Depth, + &object.CreatedAt, &object.UpdatedAt, &tombstonedAt, + ) + if err != nil { + return nil, err + } + if tombstonedAt.Valid { + object.TombstonedAt = &tombstonedAt.Time + } + return &object, nil +} diff --git a/internal/store/outbound_test.go b/internal/store/outbound_test.go new file mode 100644 --- /dev/null +++ b/internal/store/outbound_test.go @@ -0,0 +1,510 @@ +package store + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" + "tidepool/internal/testutil" +) + +// Task 14 cycle A: the outbound state the consumer writes and tasks 15-17 +// read back. These tables are the answer to a hard fact about Jetstream — a +// delete commit carries the DID, collection and rkey and NOTHING else — so +// every assertion here is really about "can a Delete still be built after the +// record is gone?". + +// outboundTestDB returns a migrated connection with task 14's tables emptied. +func outboundTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, "outbound_objects", "outbound_votes", "federation_prefs") + return database +} + +const ( + testCommentATURI = "at://" + testDID + "/social.coves.community.comment/3lzcommentaaa" + testCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + testCommunityAPID = "https://lemmy.world/c/technology" + testCommentAPID = "https://coves.social/ap/object/3lzcommentaaa" + testVoteATURI = "at://" + testDID + "/social.coves.feed.vote/3lzvoteaaaaaa" + testOtherVoteATURI = "at://" + testDID + "/social.coves.feed.vote/3lzvotebbbbbb" + testSubjectATURI = "at://" + testCommunityDID + "/social.coves.community.postv2/3lzpostaaaaaa" + testSubjectAPID = "https://lemmy.world/post/12345" +) + +func testOutboundObject() OutboundObject { + return OutboundObject{ + ATURI: testCommentATURI, + APObjectID: testCommentAPID, + LastCID: testCID, + LastRev: "3lzrev0000001", + CommunityDID: testCommunityDID, + CommunityAPID: testCommunityAPID, + TranslatedSnapshot: []byte(`{"type":"Note","content":"hello"}`), + Depth: 1, + } +} + +func testOutboundVote() OutboundVote { + return OutboundVote{ + VoteATURI: testVoteATURI, + ActorDID: testDID, + SubjectATURI: testSubjectATURI, + SubjectAPID: testSubjectAPID, + CommunityDID: testCommunityDID, + Direction: "up", + CurrentActivityID: "https://coves.social/ap/activity/" + repeatHex('a'), + } +} + +// repeatHex builds a 64-character hex-ish digest stand-in for fixtures. +func repeatHex(c byte) string { + out := make([]byte, 64) + for i := range out { + out[i] = c + } + return string(out) +} + +// --------------------------------------------------------------------------- +// outbound_objects +// --------------------------------------------------------------------------- + +func TestOutboundObjects_UpsertStartsAtSeqZeroAndBumpsOnUpdate(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + created, err := repo.Upsert(ctx, testOutboundObject()) + require.NoError(t, err, "first upsert of %s", testCommentATURI) + require.NotNil(t, created, "Upsert must return the stored row") + + assert.Equal(t, testCommentATURI, created.ATURI) + assert.Equal(t, testCommentAPID, created.APObjectID) + assert.Equal(t, testCommunityDID, created.CommunityDID) + assert.Equal(t, testCommunityAPID, created.CommunityAPID) + assert.JSONEq(t, `{"type":"Note","content":"hello"}`, string(created.TranslatedSnapshot), + "the translated snapshot is what task 15 serves the object from") + assert.Equal(t, 1, created.Depth, "reply depth rides the row (Lemmy caps comments at 50)") + assert.Equal(t, 0, created.LastActivitySeq, + "a fresh outbound object starts at seq 0: its Create is activity 0") + assert.Nil(t, created.TombstonedAt, "a created object is not tombstoned") + assert.False(t, created.IsTombstoned()) + + // An APPLIED update (the rev gate already let it through) is a second + // activity and must not reuse the Create's id. + updated := testOutboundObject() + updated.LastRev = "3lzrev0000002" + updated.LastCID = testUpdatedCID + updated.TranslatedSnapshot = []byte(`{"type":"Note","content":"edited"}`) + second, err := repo.Upsert(ctx, updated) + require.NoError(t, err, "second upsert of the same at-uri") + require.NotNil(t, second) + + assert.Equal(t, 1, second.LastActivitySeq, + "every applied write bumps last_activity_seq so each operation gets its own activity id") + assert.Equal(t, testUpdatedCID, second.LastCID, "provenance columns follow the newest commit") + assert.Equal(t, "3lzrev0000002", second.LastRev, + "last_rev is PROVENANCE ONLY — the ordering gate is jetstream_record_revs") + assert.JSONEq(t, `{"type":"Note","content":"edited"}`, string(second.TranslatedSnapshot)) + assert.Equal(t, created.CreatedAt, second.CreatedAt, "created_at is preserved across upserts") + assert.False(t, second.UpdatedAt.Before(created.UpdatedAt), "updated_at moves forward") +} + +func TestOutboundObjects_GetByATURI(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + stored, err := repo.Upsert(ctx, testOutboundObject()) + require.NoError(t, err) + require.NotNil(t, stored) + + got, err := repo.GetByATURI(ctx, testCommentATURI) + require.NoError(t, err, "GetByATURI for a stored object") + require.NotNil(t, got) + assert.Equal(t, stored.APObjectID, got.APObjectID) + assert.Equal(t, stored.CommunityAPID, got.CommunityAPID) + assert.Equal(t, stored.LastActivitySeq, got.LastActivitySeq) + + _, err = repo.GetByATURI(ctx, "at://did:plc:nobody/social.coves.community.comment/nope") + require.Error(t, err, "an unknown at-uri must not silently return a zero row") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +func TestOutboundObjects_TombstoneReturnsTheStateTheDeleteIsBuiltFrom(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + created, err := repo.Upsert(ctx, testOutboundObject()) + require.NoError(t, err) + require.NotNil(t, created) + + // The delete commit that triggers this carries no record and no CID, so + // Tombstone has to hand back everything the Delete{Note} needs. + dead, err := repo.Tombstone(ctx, testCommentATURI) + require.NoError(t, err, "tombstone %s", testCommentATURI) + require.NotNil(t, dead, "Tombstone must return the stored state, not just an error") + + require.NotNil(t, dead.TombstonedAt, "tombstoned_at must be stamped") + assert.True(t, dead.IsTombstoned()) + assert.Equal(t, testCommentAPID, dead.APObjectID, + "the AP id the Delete addresses comes from state — the delete frame has none") + assert.Equal(t, testCommunityAPID, dead.CommunityAPID, + "the community the Delete is addressed to comes from state") + assert.JSONEq(t, `{"type":"Note","content":"hello"}`, string(dead.TranslatedSnapshot), + "the snapshot survives the tombstone: task 17 restores from it") + assert.Equal(t, 1, dead.LastActivitySeq, + "the Delete is the next activity, so the seq bumps once") + + // The row SURVIVES: it is what a replayed create is rejected against. + got, err := repo.GetByATURI(ctx, testCommentATURI) + require.NoError(t, err, "a tombstoned row is still readable") + require.NotNil(t, got) + assert.True(t, got.IsTombstoned()) +} + +func TestOutboundObjects_TombstoneIsIdempotent(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundObject()) + require.NoError(t, err) + + first, err := repo.Tombstone(ctx, testCommentATURI) + require.NoError(t, err) + require.NotNil(t, first) + + second, err := repo.Tombstone(ctx, testCommentATURI) + require.NoError(t, err, "re-tombstoning is a no-op success, not an error") + require.NotNil(t, second) + + require.NotNil(t, second.TombstonedAt) + assert.Equal(t, *first.TombstonedAt, *second.TombstonedAt, + "the original tombstone time is preserved") + assert.Equal(t, first.LastActivitySeq, second.LastActivitySeq, + "a redelivered delete must reuse the activity id the first one sent, "+ + "so the seq must NOT bump again") +} + +func TestOutboundObjects_TombstoneMissingIsNotFound(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + _, err := repo.Tombstone(ctx, "at://did:plc:nobody/social.coves.community.comment/nope") + require.Error(t, err, + "a delete for a record we never federated must be distinguishable from a real tombstone") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +func TestOutboundObjects_UpsertTxRidesTheTransaction(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + // Rolled back: the outbound state must vanish with the rev-gate claim it + // rode in with, or a replay would find state but no gate row. + tx, err := database.BeginTx(ctx, nil) + require.NoError(t, err) + stored, err := repo.UpsertTx(ctx, tx, testOutboundObject()) + require.NoError(t, err, "UpsertTx inside a transaction") + require.NotNil(t, stored) + require.NoError(t, tx.Rollback()) + + _, err = repo.GetByATURI(ctx, testCommentATURI) + require.Error(t, err, "a rolled-back UpsertTx must leave no row") + assert.True(t, errors.IsNotFound(err), "want NotFound after rollback, got %v", err) + + // Committed: the row lands. + tx, err = database.BeginTx(ctx, nil) + require.NoError(t, err) + _, err = repo.UpsertTx(ctx, tx, testOutboundObject()) + require.NoError(t, err) + require.NoError(t, tx.Commit()) + + got, err := repo.GetByATURI(ctx, testCommentATURI) + require.NoError(t, err, "a committed UpsertTx must be visible") + require.NotNil(t, got) + assert.Equal(t, 0, got.LastActivitySeq, + "the rolled-back attempt must not have consumed a seq") +} + +func TestOutboundObjects_TxVariantsRejectNilTx(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + _, err := repo.UpsertTx(ctx, nil, testOutboundObject()) + require.Error(t, err, "UpsertTx with a nil tx must not silently fall back to the pool") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) + + _, err = repo.TombstoneTx(ctx, nil, testCommentATURI) + require.Error(t, err, "TombstoneTx with a nil tx must not silently fall back to the pool") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) +} + +// --------------------------------------------------------------------------- +// outbound_votes +// --------------------------------------------------------------------------- + +func TestOutboundVotes_UpsertAndBothLookups(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + created, err := repo.Upsert(ctx, testOutboundVote()) + require.NoError(t, err, "upsert vote %s", testVoteATURI) + require.NotNil(t, created, "Upsert must return the stored row") + + assert.Equal(t, testVoteATURI, created.VoteATURI) + assert.Equal(t, "up", created.Direction) + assert.Equal(t, 0, created.ActivitySeq) + assert.Equal(t, DeliveredStatePending, created.DeliveredState, + "the consumer records INTENT only: an unstated delivered_state is pending, "+ + "never delivered — task 15 claims delivery, and only on success") + + // The DELETE path's lookup: a vote delete commit carries the vote record + // at-uri and nothing else. + byURI, err := repo.GetByATURI(ctx, testVoteATURI) + require.NoError(t, err, "GetByATURI is the delete path's only key") + require.NotNil(t, byURI) + assert.Equal(t, "up", byURI.Direction, "the Undo's direction is read back from here") + assert.Equal(t, created.CurrentActivityID, byURI.CurrentActivityID, + "the Undo must embed the id the Like went out under") + + // The CREATE path's lookup: has this actor already voted here? + bySubject, err := repo.GetByActorSubject(ctx, testDID, testSubjectATURI) + require.NoError(t, err, "GetByActorSubject must find the same row") + require.NotNil(t, bySubject) + assert.Equal(t, testVoteATURI, bySubject.VoteATURI) + + _, err = repo.GetByATURI(ctx, testOtherVoteATURI) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err), "unknown vote at-uri: want NotFound, got %v", err) + + _, err = repo.GetByActorSubject(ctx, testSecondDID, testSubjectATURI) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err), "unknown (actor, subject): want NotFound, got %v", err) +} + +func TestOutboundVotes_ReUpsertBumpsSeq(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundVote()) + require.NoError(t, err) + + flipped := testOutboundVote() + flipped.Direction = "down" + second, err := repo.Upsert(ctx, flipped) + require.NoError(t, err, "re-upserting the same vote record") + require.NotNil(t, second) + + assert.Equal(t, "down", second.Direction) + assert.Equal(t, 1, second.ActivitySeq, + "a second applied write is a second activity and must not reuse the first id") +} + +func TestOutboundVotes_OneLiveVotePerActorSubject(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundVote()) + require.NoError(t, err) + + // A second vote RECORD for the same subject by the same actor: the + // unique constraint must refuse it rather than clobber the row whose + // Undo is still owed. + second := testOutboundVote() + second.VoteATURI = testOtherVoteATURI + _, err = repo.Upsert(ctx, second) + require.Error(t, err, + "a second vote record for the same (actor, subject) must not silently replace the first — "+ + "that would strand the first vote's Undo") + assert.True(t, errors.IsAlreadyExists(err), "want AlreadyExists, got %v", err) + + // The original is untouched. + got, err := repo.GetByATURI(ctx, testVoteATURI) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, 0, got.ActivitySeq, "the refused write must not have bumped the seq") +} + +func TestOutboundVotes_DeliveredStateTransitions(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundVote()) + require.NoError(t, err) + + require.NoError(t, repo.SetDeliveredState(ctx, testVoteATURI, DeliveredStateDelivered), + "task 15 flips pending -> delivered on delivery success") + got, err := repo.GetByATURI(ctx, testVoteATURI) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, DeliveredStateDelivered, got.DeliveredState) + + require.NoError(t, repo.SetDeliveredState(ctx, testVoteATURI, DeliveredStateUndone)) + got, err = repo.GetByATURI(ctx, testVoteATURI) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, DeliveredStateUndone, got.DeliveredState) + + err = repo.SetDeliveredState(ctx, testVoteATURI, DeliveredState("shipped")) + require.Error(t, err, "an unknown delivered_state must be rejected, not stored") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) + + err = repo.SetDeliveredState(ctx, testOtherVoteATURI, DeliveredStateDelivered) + require.Error(t, err, "delivering a vote we have no state for is a bug, not a no-op") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +func TestOutboundVotes_Delete(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundVote()) + require.NoError(t, err) + + require.NoError(t, repo.Delete(ctx, testVoteATURI)) + _, err = repo.GetByATURI(ctx, testVoteATURI) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err), "want NotFound after delete, got %v", err) + + require.NoError(t, repo.Delete(ctx, testVoteATURI), + "deleting an already-deleted vote is a no-op success (the callback may re-fire)") + + // With the row gone the (actor, subject) slot is free again. + fresh := testOutboundVote() + fresh.VoteATURI = testOtherVoteATURI + _, err = repo.Upsert(ctx, fresh) + require.NoError(t, err, "a re-vote after the Undo lands must be storable") +} + +func TestOutboundVotes_UpsertTxRidesTheTransaction(t *testing.T) { + database := outboundTestDB(t) + repo := NewOutboundVotes(database) + ctx := context.Background() + + tx, err := database.BeginTx(ctx, nil) + require.NoError(t, err) + _, err = repo.UpsertTx(ctx, tx, testOutboundVote()) + require.NoError(t, err) + require.NoError(t, tx.Rollback()) + + _, err = repo.GetByATURI(ctx, testVoteATURI) + require.Error(t, err, "a rolled-back UpsertTx must leave no vote state") + assert.True(t, errors.IsNotFound(err), "want NotFound after rollback, got %v", err) + + _, err = repo.UpsertTx(ctx, nil, testOutboundVote()) + require.Error(t, err, "UpsertTx with a nil tx must not silently fall back to the pool") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) +} + +// --------------------------------------------------------------------------- +// federation_prefs +// --------------------------------------------------------------------------- + +func TestFederationPrefs_AbsentRowIsNotFound(t *testing.T) { + database := outboundTestDB(t) + repo := NewFederationPrefs(database) + ctx := context.Background() + + _, err := repo.Get(ctx, testDID) + require.Error(t, err, + "federation is DEFAULT-ON and the record is an opt-out, so a user who never "+ + "spoke has no row — the store must say NotFound, never invent an enabled one") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +func TestFederationPrefs_UpsertOverwritesEveryField(t *testing.T) { + database := outboundTestDB(t) + repo := NewFederationPrefs(database) + ctx := context.Background() + + optOut := FederationPref{ + DID: testDID, + Enabled: false, + DeleteRemote: true, + Source: FederationPrefSourceRecord, + } + stored, err := repo.Upsert(ctx, optOut) + require.NoError(t, err, "upsert opt-out for %s", testDID) + require.NotNil(t, stored) + assert.False(t, stored.Enabled) + assert.True(t, stored.DeleteRemote) + assert.Equal(t, FederationPrefSourceRecord, stored.Source) + + got, err := repo.Get(ctx, testDID) + require.NoError(t, err) + require.NotNil(t, got) + assert.False(t, got.Enabled) + assert.True(t, got.DeleteRemote) + + // Re-enabling must also clear the destructive flag: a stale deleteRemote + // on an enabled row is a loaded gun pointed at task 17. + reEnabled, err := repo.Upsert(ctx, FederationPref{ + DID: testDID, + Enabled: true, + Source: FederationPrefSourceProbe, + }) + require.NoError(t, err) + require.NotNil(t, reEnabled) + assert.True(t, reEnabled.Enabled) + assert.False(t, reEnabled.DeleteRemote, + "re-enabling clears delete_remote: every field is overwritten") + assert.Equal(t, FederationPrefSourceProbe, reEnabled.Source) + assert.False(t, reEnabled.UpdatedAt.Before(stored.UpdatedAt), "updated_at moves forward") +} + +func TestFederationPrefs_SourceMustBeStated(t *testing.T) { + database := outboundTestDB(t) + repo := NewFederationPrefs(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, FederationPref{DID: testDID, Enabled: false}) + require.Error(t, err, + "the zero source is invalid: 'we read a record' and 'we went and asked' have "+ + "different staleness, and a defaulted source hides which one applied") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) + + _, err = repo.Upsert(ctx, FederationPref{ + DID: testDID, Enabled: false, Source: FederationPrefSource("guess"), + }) + require.Error(t, err, "an unknown source must be rejected") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) +} + +func TestFederationPrefs_DeleteRestoresDefaultOn(t *testing.T) { + database := outboundTestDB(t) + repo := NewFederationPrefs(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, FederationPref{ + DID: testDID, Enabled: false, Source: FederationPrefSourceRecord, + }) + require.NoError(t, err) + + require.NoError(t, repo.Delete(ctx, testDID), + "deleting the opt-out record removes the row — absence IS the default-on state") + + _, err = repo.Get(ctx, testDID) + require.Error(t, err) + assert.True(t, errors.IsNotFound(err), "want NotFound after delete, got %v", err) + + require.NoError(t, repo.Delete(ctx, testSecondDID), + "deleting a preference that never existed is a no-op success: a record delete "+ + "for a user who never opted out is the common case") +} diff --git a/internal/store/outbound_votes.go b/internal/store/outbound_votes.go new file mode 100644 --- /dev/null +++ b/internal/store/outbound_votes.go @@ -0,0 +1,161 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "tidepool/internal/errors" +) + +type postgresOutboundVotes struct { + db *sql.DB +} + +// NewOutboundVotes creates the postgres-backed outbound_votes repository. +func NewOutboundVotes(db *sql.DB) OutboundVotes { + return &postgresOutboundVotes{db: db} +} + +const outboundVoteColumns = ` + vote_at_uri, actor_did, subject_at_uri, subject_ap_id, community_did, + direction, current_activity_id, delivered_state, activity_seq, + created_at, updated_at` + +func (r *postgresOutboundVotes) Upsert(ctx context.Context, vote OutboundVote) (*OutboundVote, error) { + return r.upsert(ctx, r.db, vote) +} + +func (r *postgresOutboundVotes) UpsertTx(ctx context.Context, tx *sql.Tx, vote OutboundVote) (*OutboundVote, error) { + if tx == nil { + return nil, errors.NewValidationError("tx", "must not be nil") + } + return r.upsert(ctx, tx, vote) +} + +func (r *postgresOutboundVotes) upsert(ctx context.Context, q execer, vote OutboundVote) (*OutboundVote, error) { + // An unstated delivered_state is pending, never delivered: the consumer + // records INTENT, and only task 15 may claim delivery — on success, from + // the wire. Defaulting the zero value the other way would silently mark a + // vote as delivered that no peer ever saw, and its Undo would then look + // unnecessary. + if vote.DeliveredState == "" { + vote.DeliveredState = DeliveredStatePending + } + if !vote.DeliveredState.Valid() { + return nil, errors.NewValidationError("delivered_state", "unknown state "+string(vote.DeliveredState)) + } + + // ON CONFLICT names the PRIMARY KEY only. The (actor_did, subject_at_uri) + // constraint is deliberately NOT an upsert target: a different vote record + // for a pair that already holds one must FAIL, because overwriting the row + // would strand the Undo still owed for the first vote. + query := ` + INSERT INTO outbound_votes ( + vote_at_uri, actor_did, subject_at_uri, subject_ap_id, community_did, + direction, current_activity_id, delivered_state + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (vote_at_uri) DO UPDATE SET + actor_did = EXCLUDED.actor_did, + subject_at_uri = EXCLUDED.subject_at_uri, + subject_ap_id = EXCLUDED.subject_ap_id, + community_did = EXCLUDED.community_did, + direction = EXCLUDED.direction, + current_activity_id = EXCLUDED.current_activity_id, + delivered_state = EXCLUDED.delivered_state, + activity_seq = outbound_votes.activity_seq + 1, + updated_at = now() + RETURNING` + outboundVoteColumns + + row := q.QueryRowContext(ctx, query, + vote.VoteATURI, vote.ActorDID, vote.SubjectATURI, vote.SubjectAPID, + vote.CommunityDID, vote.Direction, vote.CurrentActivityID, string(vote.DeliveredState), + ) + stored, err := scanOutboundVote(row) + if err != nil { + // Mapped from the constraint name rather than pre-checked: a + // SELECT-then-INSERT pre-check races with a concurrent handler. + if constraint, ok := uniqueViolation(err); ok && constraint == "outbound_votes_actor_subject_key" { + return nil, errors.NewConflictError("outbound_vote", "actor_subject", vote.ActorDID+" "+vote.SubjectATURI) + } + return nil, fmt.Errorf("upsert outbound_vote %q: %w", vote.VoteATURI, err) + } + return stored, nil +} + +func (r *postgresOutboundVotes) GetByATURI(ctx context.Context, voteATURI string) (*OutboundVote, error) { + query := `SELECT` + outboundVoteColumns + ` FROM outbound_votes WHERE vote_at_uri = $1` + vote, err := scanOutboundVote(r.db.QueryRowContext(ctx, query, voteATURI)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_vote", voteATURI) + } + return nil, fmt.Errorf("get outbound_vote %q: %w", voteATURI, err) + } + return vote, nil +} + +func (r *postgresOutboundVotes) GetByActorSubject(ctx context.Context, actorDID, subjectATURI string) (*OutboundVote, error) { + query := `SELECT` + outboundVoteColumns + ` + FROM outbound_votes WHERE actor_did = $1 AND subject_at_uri = $2` + vote, err := scanOutboundVote(r.db.QueryRowContext(ctx, query, actorDID, subjectATURI)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_vote", actorDID+" on "+subjectATURI) + } + return nil, fmt.Errorf("get outbound_vote for %q on %q: %w", actorDID, subjectATURI, err) + } + return vote, nil +} + +func (r *postgresOutboundVotes) SetDeliveredState(ctx context.Context, voteATURI string, state DeliveredState) error { + // Validated in Go rather than left to the CHECK constraint: an unknown + // state is a caller bug, and the caller needs it back as a validation + // error, not as a wrapped SQLSTATE it has no way to interpret. + if !state.Valid() { + return errors.NewValidationError("delivered_state", "unknown state "+string(state)) + } + result, err := r.db.ExecContext(ctx, + `UPDATE outbound_votes SET delivered_state = $2, updated_at = now() WHERE vote_at_uri = $1`, + voteATURI, string(state)) + if err != nil { + return fmt.Errorf("set delivered_state for outbound_vote %q: %w", voteATURI, err) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("set delivered_state for outbound_vote %q: rows affected: %w", voteATURI, err) + } + if affected == 0 { + // Delivering a vote we hold no state for means the intent and the + // delivery disagree about what exists. That is a bug worth surfacing, + // not a no-op to swallow. + return errors.NewNotFoundError("outbound_vote", voteATURI) + } + return nil +} + +func (r *postgresOutboundVotes) Delete(ctx context.Context, voteATURI string) error { + // A missing row is success: the delivery callback that clears vote state + // may re-fire, and the desired end state — no live vote — already holds. + if _, err := r.db.ExecContext(ctx, + `DELETE FROM outbound_votes WHERE vote_at_uri = $1`, voteATURI); err != nil { + return fmt.Errorf("delete outbound_vote %q: %w", voteATURI, err) + } + return nil +} + +func scanOutboundVote(row rowScanner) (*OutboundVote, error) { + var vote OutboundVote + var deliveredState string + err := row.Scan( + &vote.VoteATURI, &vote.ActorDID, &vote.SubjectATURI, &vote.SubjectAPID, + &vote.CommunityDID, &vote.Direction, &vote.CurrentActivityID, + &deliveredState, &vote.ActivitySeq, &vote.CreatedAt, &vote.UpdatedAt, + ) + if err != nil { + return nil, err + } + vote.DeliveredState = DeliveredState(deliveredState) + return &vote, nil +} -- tangled.sh