| 1 | package calendar |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "time" |
| 10 | |
| 11 | "gorm.io/gorm" |
| 12 | |
| 13 | "amadan.net/oficina/calendar/internal/calendar/seal" |
| 14 | ) |
| 15 | |
| 16 | // DueReminder is everything sending one message needs, gathered in one |
| 17 | // query so the sender does no further lookups. |
| 18 | type DueReminder struct { |
| 19 | Queue ReminderQueue |
| 20 | |
| 21 | // ClaimID is the identity this worker holds on the row. Every |
| 22 | // completion passes it back and is conditional on it. |
| 23 | ClaimID string |
| 24 | |
| 25 | EventID int64 |
| 26 | Title string |
| 27 | Location string |
| 28 | Zone string |
| 29 | Address string |
| 30 | StartUTC int64 |
| 31 | |
| 32 | // DayWords is the occurrence's date written out, so the message can |
| 33 | // say "Wednesday 2 September" rather than making somebody decode a |
| 34 | // timestamp. |
| 35 | DayWords string |
| 36 | } |
| 37 | |
| 38 | // claimLease is how long a claimed reminder stays claimed before |
| 39 | // another pass may take it. |
| 40 | // |
| 41 | // It exists because a process killed between claiming and sending would |
| 42 | // otherwise leave that reminder claimed forever — never sent, and |
| 43 | // invisible. Five minutes is longer than any send takes and shorter |
| 44 | // than anybody's patience for a missing reminder. The cost of the lease |
| 45 | // expiring after a successful send that was not yet marked is one |
| 46 | // duplicate message, which is the trade this app makes on purpose. |
| 47 | const claimLease = 5 * time.Minute |
| 48 | |
| 49 | // deferFor is how long a row waits after the sender said "busy". Short: |
| 50 | // nothing was tried, and capacity comes back in seconds. |
| 51 | const deferFor = 30 * time.Second |
| 52 | |
| 53 | var errClaimLost = errors.New("calendar: claim lost") |
| 54 | |
| 55 | // newClaimID is sixteen random bytes as hex — an identity, not a secret. |
| 56 | func newClaimID() string { |
| 57 | b := make([]byte, 16) |
| 58 | if _, err := rand.Read(b); err != nil { |
| 59 | panic("calendar: no randomness for a claim id: " + err.Error()) |
| 60 | } |
| 61 | return hex.EncodeToString(b) |
| 62 | } |
| 63 | |
| 64 | // ClaimDueReminder takes the earliest claimable reminder and marks it |
| 65 | // claimed, returning everything needed to send it. |
| 66 | // |
| 67 | // The claim is one transaction: a conditional UPDATE writing a fresh |
| 68 | // claim id, then a re-read of the SAME row, and the payload is built |
| 69 | // from the re-read — never from the candidate this worker looked at |
| 70 | // first. A reschedule can land between the candidate read and the |
| 71 | // claim; the claim still succeeds on the rewritten row, and a payload |
| 72 | // built from the candidate would announce the OLD time under a valid |
| 73 | // claim (booking spec B22, r6 f2). |
| 74 | func (s *Store) ClaimDueReminder(ctx context.Context, now int64) (DueReminder, bool, error) { |
| 75 | cutoff := now - int64(claimLease.Seconds()) |
| 76 | |
| 77 | var cand ReminderQueue |
| 78 | err := s.g.WithContext(ctx). |
| 79 | Where("fired_at IS NULL AND due_at <= ?", now). |
| 80 | Where("retry_at IS NULL OR retry_at <= ?", now). |
| 81 | Where("claimed_at IS NULL OR claimed_at <= ?", cutoff). |
| 82 | Order("due_at ASC"). |
| 83 | First(&cand).Error |
| 84 | if errors.Is(err, gorm.ErrRecordNotFound) { |
| 85 | return DueReminder{}, false, nil |
| 86 | } |
| 87 | if err != nil { |
| 88 | return DueReminder{}, false, fmt.Errorf("calendar: find due reminder: %w", err) |
| 89 | } |
| 90 | |
| 91 | claimID := newClaimID() |
| 92 | var q ReminderQueue |
| 93 | err = s.g.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 94 | // The whole eligibility predicate again, due_at included: a |
| 95 | // reschedule between the candidate read and here rewrites the |
| 96 | // row to a later time, and a claim that skipped due_at would |
| 97 | // take it, send tomorrow's reminder now, and retire it. |
| 98 | res := tx.Model(&ReminderQueue{}). |
| 99 | Where("id = ? AND fired_at IS NULL AND due_at <= ?", cand.ID, now). |
| 100 | Where("retry_at IS NULL OR retry_at <= ?", now). |
| 101 | Where("claimed_at IS NULL OR claimed_at <= ?", cutoff). |
| 102 | Updates(map[string]any{"claimed_at": now, "claim_id": claimID}) |
| 103 | if res.Error != nil { |
| 104 | return fmt.Errorf("calendar: claim reminder: %w", res.Error) |
| 105 | } |
| 106 | if res.RowsAffected == 0 { |
| 107 | return errClaimLost |
| 108 | } |
| 109 | return tx.First(&q, cand.ID).Error |
| 110 | }) |
| 111 | if errors.Is(err, errClaimLost) { |
| 112 | return DueReminder{}, false, nil |
| 113 | } |
| 114 | if err != nil { |
| 115 | return DueReminder{}, false, err |
| 116 | } |
| 117 | |
| 118 | var rem Reminder |
| 119 | if err := s.g.WithContext(ctx).First(&rem, q.ReminderID).Error; err != nil { |
| 120 | return DueReminder{}, false, fmt.Errorf("calendar: read reminder: %w", err) |
| 121 | } |
| 122 | var e Event |
| 123 | err = s.g.WithContext(ctx). |
| 124 | Where("id = ? AND deleted_at IS NULL AND status <> ?", rem.EventID, StatusCancelled). |
| 125 | First(&e).Error |
| 126 | if errors.Is(err, gorm.ErrRecordNotFound) { |
| 127 | // The event went away, or was cancelled, between queueing and |
| 128 | // firing. Drop the row rather than send a reminder for |
| 129 | // something that is not happening — but only OUR row, under |
| 130 | // our claim: a reschedule that overtook us rewrote it. |
| 131 | _ = s.g.WithContext(ctx).Where("id = ? AND claim_id = ?", q.ID, claimID).Delete(&ReminderQueue{}).Error |
| 132 | return DueReminder{}, false, nil |
| 133 | } |
| 134 | if err != nil { |
| 135 | // A transient read error is not "gone". Leave the claim; the |
| 136 | // lease expires and a later pass tries again. |
| 137 | return DueReminder{}, false, fmt.Errorf("calendar: read event for reminder: %w", err) |
| 138 | } |
| 139 | var p Person |
| 140 | if err := s.g.WithContext(ctx).First(&p, rem.PersonID).Error; err != nil { |
| 141 | return DueReminder{}, false, fmt.Errorf("calendar: read person: %w", err) |
| 142 | } |
| 143 | |
| 144 | title, err := s.seal.OpenString(seal.PurposeEventTitle, e.TitleSealed) |
| 145 | if err != nil { |
| 146 | return DueReminder{}, false, fmt.Errorf("calendar: open title for reminder: %w", err) |
| 147 | } |
| 148 | loc, err := s.seal.OpenString(seal.PurposeEventLocation, e.LocationSealed) |
| 149 | if err != nil { |
| 150 | return DueReminder{}, false, fmt.Errorf("calendar: open location for reminder: %w", err) |
| 151 | } |
| 152 | |
| 153 | start := q.DueAt + int64(rem.OffsetMin)*60 |
| 154 | zone := p.Zone |
| 155 | if zone == "" { |
| 156 | zone = defaultZone |
| 157 | } |
| 158 | words := "" |
| 159 | if t, err := InZone(start, zone); err == nil { |
| 160 | words = t.Format("Monday 2 January") |
| 161 | } |
| 162 | |
| 163 | return DueReminder{ |
| 164 | Queue: q, |
| 165 | ClaimID: claimID, |
| 166 | EventID: e.ID, |
| 167 | Title: title, |
| 168 | Location: loc, |
| 169 | Zone: zone, |
| 170 | Address: p.Address, |
| 171 | StartUTC: start, |
| 172 | DayWords: words, |
| 173 | }, true, nil |
| 174 | } |
| 175 | |
| 176 | // CompleteReminder records the fire and queues the occurrence AFTER the |
| 177 | // one just sent, in one transaction conditional on the claim. |
| 178 | // |
| 179 | // Conditional, because a reschedule may have rewritten the row while |
| 180 | // this worker was sending: the row now names a different occurrence, |
| 181 | // the claim id is gone, and this worker must neither mark it fired nor |
| 182 | // advance it past a time it never announced. Zero rows means exactly |
| 183 | // that, and the worker stops; its message went out and described the |
| 184 | // old time, which the booking spec states rather than hides. |
| 185 | // |
| 186 | // LateBySec is written rather than discarded because an instance that |
| 187 | // hibernates fires late sometimes, and a product that hides that is a |
| 188 | // product whose users conclude reminders are unreliable without ever |
| 189 | // being told why. |
| 190 | func (s *Store) CompleteReminder(ctx context.Context, due DueReminder, now int64) error { |
| 191 | late := now - due.Queue.DueAt |
| 192 | if late < 0 { |
| 193 | late = 0 |
| 194 | } |
| 195 | return s.g.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 196 | res := tx.Model(&ReminderQueue{}). |
| 197 | Where("id = ? AND claim_id = ?", due.Queue.ID, due.ClaimID). |
| 198 | Updates(map[string]any{"fired_at": now, "late_by_sec": late}) |
| 199 | if res.Error != nil { |
| 200 | return fmt.Errorf("calendar: mark reminder sent: %w", res.Error) |
| 201 | } |
| 202 | if res.RowsAffected == 0 { |
| 203 | return nil |
| 204 | } |
| 205 | // Queue the occurrence AFTER the one just sent. Not "the next |
| 206 | // occurrence from now": a reminder fires before its event |
| 207 | // starts, so that search returns the one just sent and the |
| 208 | // drain re-sends it until its deadline. |
| 209 | return s.rescheduleAfterTx(tx, due.Queue.ReminderID, due.StartUTC) |
| 210 | }) |
| 211 | } |
| 212 | |
| 213 | // DeferReminder puts a row back without counting an attempt: the sender |
| 214 | // was busy and nothing was tried. retry_at, never due_at — due_at IS |
| 215 | // the meeting time minus the offset. |
| 216 | func (s *Store) DeferReminder(ctx context.Context, queueID int64, claimID string, now int64) (ok bool, err error) { |
| 217 | res := s.g.WithContext(ctx).Model(&ReminderQueue{}). |
| 218 | Where("id = ? AND claim_id = ?", queueID, claimID). |
| 219 | Updates(map[string]any{"retry_at": now + int64(deferFor.Seconds()), "claimed_at": nil, "claim_id": nil}) |
| 220 | if res.Error != nil { |
| 221 | return false, fmt.Errorf("calendar: defer reminder: %w", res.Error) |
| 222 | } |
| 223 | return res.RowsAffected == 1, nil |
| 224 | } |
| 225 | |
| 226 | // MaxReminderAttempts is when a reminder stops being retried. |
| 227 | // |
| 228 | // Some addresses are permanently undeliverable, and retrying one |
| 229 | // forever is a queue that never drains and an instance that never |
| 230 | // hibernates — which costs real money on a platform billed by |
| 231 | // residency. |
| 232 | const MaxReminderAttempts = 5 |
| 233 | |
| 234 | // FailReminder records an attempt and gives up after enough of them, |
| 235 | // conditional on the claim. |
| 236 | // |
| 237 | // occurrenceStartUTC is the start of the occurrence that just failed, |
| 238 | // needed only on the giving-up path so the NEXT one can be queued. |
| 239 | // |
| 240 | // The claim is REFRESHED rather than cleared. Clearing it put the row |
| 241 | // straight back at the head of ClaimDueReminder's `ORDER BY due_at |
| 242 | // ASC`, so the very next claim in the same pass returned the same row, |
| 243 | // the drain's `seen` guard tripped, and the pass returned early — with |
| 244 | // every other due reminder still unsent. Leaving the lease in place |
| 245 | // makes the five-minute claim do the backoff it was already written to |
| 246 | // do, and the pass moves on to the next person's reminder. |
| 247 | // |
| 248 | // Giving up RESCHEDULES the rule. See Reminder.UndeliverableAt: the |
| 249 | // marker lives on the rule so the queue row is free to move on. |
| 250 | func (s *Store) FailReminder(ctx context.Context, queueID int64, claimID string, occurrenceStartUTC int64) error { |
| 251 | var q ReminderQueue |
| 252 | if err := s.g.WithContext(ctx).First(&q, queueID).Error; err != nil { |
| 253 | return fmt.Errorf("calendar: read queue row: %w", err) |
| 254 | } |
| 255 | now := time.Now().Unix() |
| 256 | attempts := q.Attempts + 1 |
| 257 | |
| 258 | return s.g.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 259 | res := tx.Model(&ReminderQueue{}). |
| 260 | Where("id = ? AND claim_id = ?", queueID, claimID). |
| 261 | Updates(map[string]any{"attempts": attempts, "claimed_at": now}) |
| 262 | if res.Error != nil { |
| 263 | return fmt.Errorf("calendar: record reminder failure: %w", res.Error) |
| 264 | } |
| 265 | if res.RowsAffected == 0 || attempts < MaxReminderAttempts { |
| 266 | return nil |
| 267 | } |
| 268 | if err := tx.Model(&Reminder{}). |
| 269 | Where("id = ?", q.ReminderID). |
| 270 | Update("undeliverable_at", now).Error; err != nil { |
| 271 | return fmt.Errorf("calendar: mark reminder undeliverable: %w", err) |
| 272 | } |
| 273 | return s.rescheduleAfterTx(tx, q.ReminderID, occurrenceStartUTC) |
| 274 | }) |
| 275 | } |
| 276 | |
| 277 | // UndeliverableReminders lists the ones that gave up, for the settings |
| 278 | // page. |
| 279 | // Counted off the RULE, not the queue row. The queue row is rescheduled |
| 280 | // the moment a rule gives up, so its attempts counter is reset and would |
| 281 | // report nothing was ever undeliverable. |
| 282 | func (s *Store) UndeliverableReminders(ctx context.Context, personID int64) (int64, error) { |
| 283 | var n int64 |
| 284 | err := s.g.WithContext(ctx).Model(&Reminder{}). |
| 285 | Where("person_id = ? AND undeliverable_at IS NOT NULL", personID). |
| 286 | Count(&n).Error |
| 287 | if err != nil { |
| 288 | return 0, fmt.Errorf("calendar: count undeliverable: %w", err) |
| 289 | } |
| 290 | return n, nil |
| 291 | } |
| 292 | |
amadan