| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | |
| 7 | "github.com/carlosframework/platform/internal/hostops" |
| 8 | "github.com/carlosframework/platform/internal/release" |
| 9 | ) |
| 10 | |
| 11 | // The mail tables (mail_sender_counts, mail_events, mail_event_hooks, |
| 12 | // mail_event_deliveries) key an app by its account SQID, not the account id |
| 13 | // the generated adapters take, so they get this one adapter instead. Every |
| 14 | // write to them goes through it, scoped to exactly the app it writes: |
| 15 | // |
| 16 | // - Inside a context that already holds a permit for the app (a door's |
| 17 | // release.WithAppMutation / Server.claimAppMutation, or a mail |
| 18 | // admission's context), RunSQLIngest joins it: one operation for the |
| 19 | // whole door, never one per statement. |
| 20 | // - Otherwise (the SES ingest, the push worker) it opens that app's own |
| 21 | // shared SQL ingest registration around the write. The non-expiring |
| 22 | // mail delivery operation itself is unchanged. |
| 23 | // |
| 24 | // A deployment-wide statement (the retention prunes) is split per app from a |
| 25 | // fresh read of the rows it would touch, so each app's rows go under that |
| 26 | // app's permit and no write is ever unscoped. |
| 27 | // |
| 28 | // Flag-off (no SQL gate) and inside an already bound view, call runs on s |
| 29 | // unchanged, exactly as before. |
| 30 | func (s *Store) withMailAppSQL(ctx context.Context, sqid, app, action string, call func(*Store) error) error { |
| 31 | if s.gate == nil || s.bound != nil { |
| 32 | return call(s) |
| 33 | } |
| 34 | if s.outerInsideView() { |
| 35 | return errOuterStoreInView |
| 36 | } |
| 37 | if !release.ValidAccount(sqid) || !release.ValidApp(app) { |
| 38 | return hostops.ErrIdentity |
| 39 | } |
| 40 | acct, err := s.AccountBySqid(sqid) |
| 41 | if err != nil { |
| 42 | return err |
| 43 | } |
| 44 | if acct == nil { |
| 45 | return hostops.ErrIdentity |
| 46 | } |
| 47 | return s.withAutoAppSQL(ctx, acct.ID, app, action, call) |
| 48 | } |
| 49 | |
| 50 | type mailAppKey struct{ Account, App string } |
| 51 | |
| 52 | // mailAppsWith lists the distinct apps with rows matching where in table: the |
| 53 | // affected app set of a deployment-wide statement. The rows are closed |
| 54 | // before any write starts (the gated store has one connection). |
| 55 | func (s *Store) mailAppsWith(ctx context.Context, table, where string, args ...any) ([]mailAppKey, error) { |
| 56 | rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT account, app FROM `+table+` WHERE `+where+` ORDER BY account, app`, args...) |
| 57 | if err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | defer rows.Close() |
| 61 | var out []mailAppKey |
| 62 | for rows.Next() { |
| 63 | var k mailAppKey |
| 64 | if err := rows.Scan(&k.Account, &k.App); err != nil { |
| 65 | return nil, err |
| 66 | } |
| 67 | out = append(out, k) |
| 68 | } |
| 69 | return out, rows.Err() |
| 70 | } |
| 71 | |
| 72 | // pruneMailAppRows deletes table's rows with col < cutoff one app at a time, |
| 73 | // each under that app's own mutation. A refused app (an operation holds it, |
| 74 | // or its account is gone) keeps its rows until a later pass; the others |
| 75 | // still go. |
| 76 | func (s *Store) pruneMailAppRows(ctx context.Context, table, col string, cutoff int64, action string) error { |
| 77 | apps, err := s.mailAppsWith(ctx, table, col+` < ?`, cutoff) |
| 78 | if err != nil { |
| 79 | return err |
| 80 | } |
| 81 | var errs []error |
| 82 | for _, k := range apps { |
| 83 | k := k |
| 84 | errs = append(errs, s.withMailAppSQL(ctx, k.Account, k.App, action, func(v *Store) error { |
| 85 | _, err := v.db.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ? AND `+col+` < ?`, k.Account, k.App, cutoff) |
| 86 | return err |
| 87 | })) |
| 88 | } |
| 89 | return errors.Join(errs...) |
| 90 | } |
| 91 | |
amadan