Something went wrong. Try again.
An event companion and onboarding experience for the ATmosphere (alpha) atmo.quest
Something went wrong. Try again.
4.5 kB · 144 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145package connection
import ( "context" "database/sql" "errors" "fmt"
"github.com/bluesky-social/indigo/atproto/syntax")
// PendingItem is a queued reciprocal connection that needs to be written to// target_did's PDS the next time they log in.type PendingItem struct { ID int64 TargetDID syntax.DID InitiatorDID syntax.DID EventURI string}
// Queue is a SQLite-backed FIFO of reciprocal connection writes. Methods are// safe for concurrent callers (SQLite's own write lock serializes).type Queue struct { db *sql.DB}
// NewQueue constructs a Queue against the provided database. The caller is// responsible for running migrations (003_pending_connections.sql).func NewQueue(db *sql.DB) *Queue { return &Queue{db: db}}
// Enqueue records that `target` owes a connection record listing `initiator`.// Idempotent: re-enqueueing the same (target, initiator, eventURI) is a no-op// thanks to the unique index.func (q *Queue) Enqueue(ctx context.Context, target, initiator syntax.DID, eventURI string) error { if target == "" || initiator == "" { return errors.New("connection queue: empty DID") } if target == initiator { return errors.New("connection queue: self-connection") } _, err := q.db.ExecContext(ctx, ` INSERT INTO pending_connections (target_did, initiator_did, event_uri) VALUES (?, ?, ?) ON CONFLICT(target_did, initiator_did, event_uri) DO NOTHING `, target.String(), initiator.String(), eventURI) if err != nil { return fmt.Errorf("enqueue: %w", err) } return nil}
// EnqueueLocal records that an ATProto target owes a reciprocal connection// to a local user. After the local user graduates, these rows are migrated// to the main pending_connections table.func (q *Queue) EnqueueLocal(ctx context.Context, target syntax.DID, initiatorLocalID, eventURI string) error { if target == "" || initiatorLocalID == "" { return errors.New("connection queue: empty target DID or initiator local ID") } _, err := q.db.ExecContext(ctx, ` INSERT INTO pending_local_connections (target_did, initiator_local_id, event_uri) VALUES (?, ?, ?) ON CONFLICT(target_did, initiator_local_id, event_uri) DO NOTHING `, target.String(), initiatorLocalID, eventURI) return err}
// List returns up to `limit` pending items for the given target. If limit is// <= 0, all pending items are returned.func (q *Queue) List(ctx context.Context, target syntax.DID, limit int) ([]PendingItem, error) { var ( rows *sql.Rows err error ) if limit <= 0 { rows, err = q.db.QueryContext(ctx, ` SELECT id, target_did, initiator_did, event_uri FROM pending_connections WHERE target_did = ? ORDER BY id `, target.String()) } else { rows, err = q.db.QueryContext(ctx, ` SELECT id, target_did, initiator_did, event_uri FROM pending_connections WHERE target_did = ? ORDER BY id LIMIT ? `, target.String(), limit) } if err != nil { return nil, fmt.Errorf("list pending: %w", err) } defer rows.Close()
var out []PendingItem for rows.Next() { var ( id int64 targetStr string initiatorStr string eventURI string ) if err := rows.Scan(&id, &targetStr, &initiatorStr, &eventURI); err != nil { return nil, fmt.Errorf("scan pending: %w", err) } tDID, err := syntax.ParseDID(targetStr) if err != nil { return nil, fmt.Errorf("pending target DID parse: %w", err) } iDID, err := syntax.ParseDID(initiatorStr) if err != nil { return nil, fmt.Errorf("pending initiator DID parse: %w", err) } out = append(out, PendingItem{ID: id, TargetDID: tDID, InitiatorDID: iDID, EventURI: eventURI}) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate pending: %w", err) } return out, nil}
// Delete removes a single pending row by its primary key. Used after a// successful PDS write during Drain.func (q *Queue) Delete(ctx context.Context, id int64) error { _, err := q.db.ExecContext(ctx, `DELETE FROM pending_connections WHERE id = ?`, id) if err != nil { return fmt.Errorf("delete pending %d: %w", id, err) } return nil}
// Count returns the number of pending items for the given target.func (q *Queue) Count(ctx context.Context, target syntax.DID) (int, error) { var n int err := q.db.QueryRowContext(ctx, ` SELECT COUNT(*) FROM pending_connections WHERE target_did = ? `, target.String()).Scan(&n) if err != nil { return 0, fmt.Errorf("count pending: %w", err) } return n, nil}