diff --git a/deliberi/account_ingest_test.go b/deliberi/account_ingest_test.go new file mode 100644 index 000000000..03574064c --- /dev/null +++ b/deliberi/account_ingest_test.go @@ -0,0 +1,65 @@ +package deliberi + +import ( + "slices" + "testing" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + deldb "tangled.org/core/deliberi/db" + models "tangled.org/core/deliberi/models" + "tangled.org/core/jetstream" +) + +type kvDeletes struct{ keys []string } + +func (k *kvDeletes) PutEmailDid(string, string) {} +func (k *kvDeletes) SetPrimaryEmail(string, string) {} +func (k *kvDeletes) DeleteEmailKey(email string) { k.keys = append(k.keys, email) } +func (k *kvDeletes) DeletePrimaryEmail(did string) { k.keys = append(k.keys, "__primary:"+did) } + +func accountFrame(did string, active bool, status string) *jetstream.Event { + frameClock++ + acct := &comatproto.SyncSubscribeRepos_Account{Did: did, Active: active} + if status != "" { + acct.Status = &status + } + return &jetstream.Event{Did: did, TimeUS: frameClock, Kind: jetstream.EventKindAccount, Account: acct} +} + +func TestAccountDeletionDropsEmails(t *testing.T) { + const gone, kept = "did:plc:gone", "did:plc:kept" + i := newTestIngester(t, fakeBobbin{}) + rec := &kvDeletes{} + i.emailKV = rec + for _, e := range []models.Email{ + {Did: gone, Address: "gone@example.com", Verified: true, Primary: true}, + {Did: gone, Address: "alt@example.com", Verified: true}, + {Did: kept, Address: "kept@example.com", Verified: true, Primary: true}, + } { + if err := deldb.AddEmail(i.db, e); err != nil { + t.Fatal(err) + } + } + + mustProcess(t, i, + accountFrame(gone, false, "deactivated"), + accountFrame(kept, false, "takendown"), + ) + if len(rec.keys) != 0 { + t.Fatalf("reversible statuses touched KV: %v", rec.keys) + } + + mustProcess(t, i, accountFrame(gone, false, "deleted")) + + if left, _ := deldb.GetAllEmails(i.db, gone); len(left) != 0 { + t.Errorf("deleted account still has emails: %v", left) + } + if left, _ := deldb.GetAllEmails(i.db, kept); len(left) != 1 { + t.Errorf("other account lost emails: %v", left) + } + slices.Sort(rec.keys) + want := []string{"__primary:" + gone, "alt@example.com", "gone@example.com"} + if !slices.Equal(rec.keys, want) { + t.Errorf("KV deletes = %v, want %v", rec.keys, want) + } +} diff --git a/deliberi/db/email.go b/deliberi/db/email.go index a32fece72..0715ef8a5 100644 --- a/deliberi/db/email.go +++ b/deliberi/db/email.go @@ -237,6 +237,23 @@ func DeleteEmail(e Execer, did string, email string) error { return err } +func DeleteAccountEmails(e Execer, did string) ([]string, error) { + rows, err := e.Query(`delete from emails where did = ? returning email`, did) + if err != nil { + return nil, err + } + defer rows.Close() + var emails []string + for rows.Next() { + var email string + if err := rows.Scan(&email); err != nil { + return nil, err + } + emails = append(emails, email) + } + return emails, rows.Err() +} + func MarkEmailVerified(e Execer, did string, email string) error { query := ` update emails diff --git a/deliberi/deliberi.go b/deliberi/deliberi.go index d3cfa709b..5c54b70da 100644 --- a/deliberi/deliberi.go +++ b/deliberi/deliberi.go @@ -34,9 +34,16 @@ func Run(ctx context.Context, cfg *config.Config) error { events := analytics.New(cfg.Posthog, log.SubLogger(logger, "analytics")) defer events.Close() + var emailKV kv.EmailKV + if cfg.KvApiToken != "" && cfg.CloudflareAccountID != "" && cfg.KvNamespaceID != "" { + emailKV = kv.NewClient(cfg.KvApiToken, cfg.CloudflareAccountID, cfg.KvNamespaceID, + log.SubLogger(logger, "kv")) + } + bobbin := newBobbinClient(cfg.BobbinApiUrl) ingester := NewIngester(database, bobbin, resolver, cfg.JetstreamEndpoint, log.SubLogger(logger, "ingest")) ingester.analytics = events + ingester.emailKV = emailKV ingestDone := make(chan struct{}) go func() { defer close(ingestDone) @@ -60,11 +67,7 @@ func Run(ctx context.Context, cfg *config.Config) error { Sender: sender, Invites: NewInviteSync(database, bobbin, log.SubLogger(logger, "invites")), Analytics: events, - } - - if cfg.KvApiToken != "" && cfg.CloudflareAccountID != "" && cfg.KvNamespaceID != "" { - x.KV = kv.NewClient(cfg.KvApiToken, cfg.CloudflareAccountID, cfg.KvNamespaceID, - log.SubLogger(logger, "kv")) + KV: emailKV, } srv := &http.Server{Addr: cfg.ListenAddr, Handler: chiMount(x)} diff --git a/deliberi/ingest.go b/deliberi/ingest.go index 80aae69c9..ed6e58968 100644 --- a/deliberi/ingest.go +++ b/deliberi/ingest.go @@ -12,6 +12,7 @@ import ( "github.com/posthog/posthog-go" "tangled.org/core/deliberi/analytics" deldb "tangled.org/core/deliberi/db" + "tangled.org/core/deliberi/kv" models "tangled.org/core/deliberi/models" "tangled.org/core/idresolver" "tangled.org/core/jetstream" @@ -33,6 +34,8 @@ type Ingester struct { logger *slog.Logger // nil when analytics is off analytics *analytics.Client + // nil when the KV cache is off + emailKV kv.EmailKV idleCursorUs int64 } @@ -88,6 +91,9 @@ func (i *Ingester) Run(ctx context.Context) error { type writer func(x deldb.Execer) error func (i *Ingester) process(ctx context.Context, e *jetstream.Event) error { + if e.Kind == jetstream.EventKindAccount && e.Account != nil { + return i.processAccount(e) + } var write writer if e.Kind == jetstream.EventKindCommit && e.Commit != nil { write = i.plan(ctx, e) @@ -111,6 +117,38 @@ func (i *Ingester) process(ctx context.Context, e *jetstream.Event) error { return tx.Commit() } +// only "deleted" is final, deactivated and takendown accounts can come back +func (i *Ingester) processAccount(e *jetstream.Event) error { + if e.Account.Active || e.Account.Status == nil || *e.Account.Status != "deleted" { + return i.advance(e.TimeUS) + } + tx, err := i.db.Begin() + if err != nil { + return fmt.Errorf("starting account tx: %w", err) + } + defer tx.Rollback() + emails, err := deldb.DeleteAccountEmails(tx, e.Did) + if err != nil { + return fmt.Errorf("deleting emails for %s: %w", e.Did, err) + } + if err := deldb.SaveCursor(tx, e.TimeUS); err != nil { + return fmt.Errorf("saving cursor: %w", err) + } + if err := tx.Commit(); err != nil { + return err + } + if len(emails) > 0 { + i.logger.Info("removed emails of deleted account", "did", e.Did, "count", len(emails)) + } + if i.emailKV != nil { + for _, email := range emails { + i.emailKV.DeleteEmailKey(email) + } + i.emailKV.DeletePrimaryEmail(e.Did) + } + return nil +} + func (i *Ingester) advance(timeUs int64) error { if timeUs-i.idleCursorUs < idleCursorEvery.Microseconds() { return nil diff --git a/deliberi/kv/kv.go b/deliberi/kv/kv.go index eaaf92dbb..1def6c24d 100644 --- a/deliberi/kv/kv.go +++ b/deliberi/kv/kv.go @@ -15,6 +15,7 @@ type EmailKV interface { PutEmailDid(email, did string) SetPrimaryEmail(did, email string) DeleteEmailKey(email string) + DeletePrimaryEmail(did string) } type Client struct { @@ -51,6 +52,10 @@ func (c *Client) DeleteEmailKey(email string) { c.delete("delete email key", strings.ToLower(email)) } +func (c *Client) DeletePrimaryEmail(did string) { + c.delete("delete primary email", "__primary:"+did) +} + func (c *Client) write(action, key, value string) { if c == nil { return diff --git a/deliberi/xrpc/account_test.go b/deliberi/xrpc/account_test.go index f7d3c2202..f5bf91b26 100644 --- a/deliberi/xrpc/account_test.go +++ b/deliberi/xrpc/account_test.go @@ -45,6 +45,12 @@ func (k *kvRecorder) DeleteEmailKey(email string) { k.deletes = append(k.deletes, email) } +func (k *kvRecorder) DeletePrimaryEmail(did string) { + k.mu.Lock() + defer k.mu.Unlock() + k.deletes = append(k.deletes, "__primary:"+did) +} + func (k *kvRecorder) snapshot() (emailDids, primaries, deletes []string) { k.mu.Lock() defer k.mu.Unlock()