| 1 | package calls |
| 2 | |
| 3 | import ( |
| 4 | "amadan.net/oficina/estilo.git/chrome" |
| 5 | "amadan.net/oficina/estilo.git/sky" |
| 6 | "amadan.net/oficina/estilo.git/theme" |
| 7 | "context" |
| 8 | "crypto/rand" |
| 9 | "encoding/hex" |
| 10 | "fmt" |
| 11 | "log/slog" |
| 12 | "net/http" |
| 13 | "sync/atomic" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/go-chi/chi/v5" |
| 17 | gormlogger "gorm.io/gorm/logger" |
| 18 | |
| 19 | rauth "amadan.net/rastrillo/rastrillo/auth" |
| 20 | "amadan.net/rastrillo/rastrillo/csrf" |
| 21 | "amadan.net/rastrillo/rastrillo/db" |
| 22 | "amadan.net/rastrillo/rastrillo/migrate" |
| 23 | |
| 24 | "amadan.net/oficina/calls/internal/calls/apns" |
| 25 | "amadan.net/oficina/calls/internal/calls/hub" |
| 26 | "amadan.net/oficina/calls/internal/calls/relay" |
| 27 | "amadan.net/oficina/calls/internal/calls/seal" |
| 28 | ) |
| 29 | |
| 30 | // app holds what every handler needs. |
| 31 | type app struct { |
| 32 | // signinMail is the queue sign-in links go through (signin_mail.go). |
| 33 | signinMail *mailQueue |
| 34 | db *db.DB |
| 35 | cfg resolved |
| 36 | sealer *seal.Sealer |
| 37 | store *store |
| 38 | logger *slog.Logger |
| 39 | auth *rauth.Auth |
| 40 | home *homeIntegration |
| 41 | nativeAuth *nativeAuthorization |
| 42 | linkBudget *budget |
| 43 | outsiderBudget *budget |
| 44 | // overBudget counts outsider sign-in attempts the budgets refused, |
| 45 | // for sweepHourly to report: never logged on the response path. |
| 46 | overBudget atomic.Int64 |
| 47 | hub *hub.Hub |
| 48 | // push is the APNs sender, or nil where no key is configured — which |
| 49 | // is every test and every deployment that has not been given one. |
| 50 | // Nil is not a degraded mode: knocks still reach every member who is |
| 51 | // in the room, exactly as before. It is only the member who is not |
| 52 | // there who goes untold. |
| 53 | push knockPusher |
| 54 | // hubSalt keys the session cookie into the hub's session id, so the |
| 55 | // hub holds no session material it did not have to. |
| 56 | hubSalt string |
| 57 | // signalLimiter is POST /m/{sqid}/signal's per-peer token bucket |
| 58 | // (spec §4.4): thirty a second, burst sixty. |
| 59 | signalLimiter *rateLimiter |
| 60 | // apiLimiter is the drive surface's per-agent token bucket (spec |
| 61 | // §8): sixty requests a minute, burst sixty, keyed "agent:<id>". |
| 62 | apiLimiter *rateLimiter |
| 63 | // qualityLimiter and problemLimiter are the report endpoints' (spec |
| 64 | // §15.26), keyed on the reporting participation's tag: a quality |
| 65 | // report every ten seconds, burst six; a problem report a minute, |
| 66 | // burst three. |
| 67 | qualityLimiter *rateLimiter |
| 68 | problemLimiter *rateLimiter |
| 69 | // reportLog is where the reports go: Config.ReportLogger. |
| 70 | reportLog *slog.Logger |
| 71 | |
| 72 | // tickPass is the housekeeping pass tickLoop runs, nil in production |
| 73 | // and everywhere but the one test that records what deadline the |
| 74 | // scheduling path hands it (see tickLoop). |
| 75 | tickPass func(ctx context.Context, within time.Duration) |
| 76 | |
| 77 | // done closes when the context App was given is cancelled — a |
| 78 | // SIGTERM, in the deployed shape. Live streams watch it and return, |
| 79 | // because http.Server.Shutdown waits for handlers without |
| 80 | // cancelling their contexts: one open stream would otherwise hold |
| 81 | // the whole shutdown deadline. |
| 82 | done <-chan struct{} |
| 83 | |
| 84 | // The stream's two timings, held as atomics because a test moves |
| 85 | // them (SetStreamTimings) while stream handlers are reading them. |
| 86 | heartbeat atomic.Int64 |
| 87 | ttl atomic.Int64 |
| 88 | // iceLifeTest, when set (tests, development only), is the life every |
| 89 | // /ice answer is minted with (SetIceLife). |
| 90 | iceLifeTest atomic.Int64 |
| 91 | |
| 92 | // allow is the team, held as an atomic for the same reason: a test |
| 93 | // drops a member from it (SetAllowlist) while the tick goroutine is |
| 94 | // re-deriving standing. Nil means "as resolved at boot"; nothing in |
| 95 | // production ever stores into it. Read it with allowNow(). |
| 96 | allow atomic.Pointer[allowlist] |
| 97 | |
| 98 | // relay is the parsed CARLOS_RELAY contract, held as an atomic so a |
| 99 | // test can hand the app a relay after boot (TestHandle.SetRelay) |
| 100 | // while a request is reading it. build stores rc.relay once; |
| 101 | // production never stores again. Read it with relayRegions(). |
| 102 | relay atomic.Value |
| 103 | } |
| 104 | |
| 105 | // relayRegions is the parsed contract, swapped atomically so a test can |
| 106 | // hand the app a relay after boot (TestHandle.SetRelay). Production |
| 107 | // sets it once in build. |
| 108 | func (a *app) relayRegions() []relay.Region { |
| 109 | v, _ := a.relay.Load().([]relay.Region) |
| 110 | return v |
| 111 | } |
| 112 | |
| 113 | func (a *app) relayConfigured() bool { return len(a.relayRegions()) > 0 } |
| 114 | |
| 115 | func (a *app) heartbeatEvery() time.Duration { return atLeast(a.heartbeat.Load()) } |
| 116 | func (a *app) streamLife() time.Duration { return atLeast(a.ttl.Load()) } |
| 117 | |
| 118 | // atLeast keeps a timing above the floor: time.NewTicker panics on a |
| 119 | // zero or negative duration, and that would be inside a request. |
| 120 | func atLeast(ns int64) time.Duration { |
| 121 | if d := time.Duration(ns); d > minStreamTiming { |
| 122 | return d |
| 123 | } |
| 124 | return minStreamTiming |
| 125 | } |
| 126 | |
| 127 | // TestHandle is the test-only control surface NewForTest hands back: |
| 128 | // the app's clock, and the sweep that would otherwise only run on a |
| 129 | // timer. Tests only; nothing in production holds one. |
| 130 | type TestHandle struct { |
| 131 | a *app |
| 132 | stop context.CancelFunc |
| 133 | } |
| 134 | |
| 135 | // ShutdownForTest cancels the context the app was built on, which is |
| 136 | // what a SIGTERM does to the deployed process: live streams see it and |
| 137 | // return. It does not stop the handler serving requests. |
| 138 | func (h *TestHandle) ShutdownForTest() { h.stop() } |
| 139 | |
| 140 | // SetStreamTimings shortens the SSE heartbeat and TTL so a test can |
| 141 | // watch a stream expire without waiting five minutes. Set it before any |
| 142 | // stream opens: a stream reads both once, when it starts. |
| 143 | func (h *TestHandle) SetStreamTimings(heartbeat, ttl time.Duration) { |
| 144 | h.a.heartbeat.Store(int64(heartbeat)) |
| 145 | h.a.ttl.Store(int64(ttl)) |
| 146 | } |
| 147 | |
| 148 | // SetIceLife shortens the relay credentials /ice mints, for every request |
| 149 | // (the X-Calls-Test-Ice-TTL header does it for one): a gate that needs a |
| 150 | // renewal mid-call cannot add a header to the native core's fetch. Zero |
| 151 | // puts the production life back. Tests only; development only. |
| 152 | func (h *TestHandle) SetIceLife(d time.Duration) { h.a.iceLifeTest.Store(int64(d)) } |
| 153 | |
| 154 | // SetAllowlist replaces the team, so a test can drop a member and watch |
| 155 | // the next tick apply spec §2.4's transition to everyone that member |
| 156 | // holds a participation for. The addresses are normalized exactly as |
| 157 | // resolveAllowlist normalizes $CALLS_ALLOWED_ADDRESSES, and the list is |
| 158 | // always non-nil: SetAllowlist() with no addresses means "nobody is a |
| 159 | // member", never development's "everybody is". |
| 160 | func (h *TestHandle) SetAllowlist(addrs ...string) { |
| 161 | list := allowlist{addrs: map[string]struct{}{}} |
| 162 | for _, addr := range addrs { |
| 163 | if addr = normalizeAddress(addr); addr != "" { |
| 164 | list.addrs[addr] = struct{}{} |
| 165 | } |
| 166 | } |
| 167 | h.a.allow.Store(&list) |
| 168 | } |
| 169 | |
| 170 | // SetRelay hands the app a CARLOS_RELAY value after boot. Tests only. |
| 171 | func (h *TestHandle) SetRelay(raw string) error { |
| 172 | regions, err := relay.Parse(raw) |
| 173 | if err != nil { |
| 174 | return err |
| 175 | } |
| 176 | h.a.relay.Store(regions) |
| 177 | return nil |
| 178 | } |
| 179 | |
| 180 | // TickForTest runs one pass of the hub housekeeping App runs every |
| 181 | // fifteen seconds: unjoined and lapsed participations dropped, |
| 182 | // tombstones expired, every joined kind re-derived from standing. |
| 183 | func (h *TestHandle) TickForTest() { |
| 184 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 185 | defer cancel() |
| 186 | h.a.hub.Tick(ctx) |
| 187 | } |
| 188 | |
| 189 | // SetClock replaces the store's clock, so a test can expire grants, |
| 190 | // pending rows and magic links without sleeping. |
| 191 | func (h *TestHandle) SetClock(now func() time.Time) { h.a.store.clk.set(now) } |
| 192 | |
| 193 | // SweepForTest runs one bounded pass of what the hourly ticker runs, |
| 194 | // using now for both store.sweep and auth.Sweep — Internal 5's one |
| 195 | // clock for the whole pass, given explicitly rather than read back off |
| 196 | // the store, so a test does not have to keep the two in step by hand. |
| 197 | func (h *TestHandle) SweepForTest(now time.Time) { |
| 198 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 199 | defer cancel() |
| 200 | h.a.sweepOnce(ctx, now) |
| 201 | } |
| 202 | |
| 203 | // NextDue answers what Runtime.NextDue would answer right now — the |
| 204 | // platform wake-up time App wires to Options.NextDue. Exposed so a |
| 205 | // black-box callstest test can check it after driving a real request, |
| 206 | // without reaching into the store directly. |
| 207 | func (h *TestHandle) NextDue() time.Time { return h.a.nextDue() } |
| 208 | |
| 209 | // SignalLimiterSize answers how many keys the signal endpoint's rate |
| 210 | // limiter currently holds — a black-box test's only window into |
| 211 | // whether a request minted a bucket at all (review Task 5 item 1: a |
| 212 | // room id rotation, or anything else refused before the limiter, must |
| 213 | // leave this unchanged). |
| 214 | func (h *TestHandle) SignalLimiterSize() int { return h.a.signalLimiter.size() } |
| 215 | |
| 216 | // gormWriter is GORM's logger.Writer over the app's slog logger, so |
| 217 | // GORM's warnings land in one log stream rather than a second stdout |
| 218 | // format. rastrillo/db has the same adapter, unexported. |
| 219 | type gormWriter struct{ log *slog.Logger } |
| 220 | |
| 221 | func (w gormWriter) Printf(format string, args ...any) { |
| 222 | w.log.Warn(fmt.Sprintf(format, args...)) |
| 223 | } |
| 224 | |
| 225 | // sweepEvery is how often expired rows are cleared. Nothing depends on |
| 226 | // it for correctness — every read filters on expiry — so the interval |
| 227 | // is about not keeping records past their life, not about safety. |
| 228 | const sweepEvery = time.Hour |
| 229 | |
| 230 | // sweepOnce clears Calls' expired pending rows and rastrillo's expired |
| 231 | // auth links and sessions (review C1-7), one bounded call to |
| 232 | // store.sweep — not looped, so a caller with a deadline (the boot sweep, |
| 233 | // SweepForTest) gets one page of work done and returns. An unclicked |
| 234 | // magic link holds a plaintext address in auth_links, and rastrillo only |
| 235 | // deletes it when the token is presented or when Sweep is called; nobody |
| 236 | // was calling Sweep. |
| 237 | // |
| 238 | // now drives both halves: one clock for the whole pass (Internal 5), so |
| 239 | // a test that advances Calls' clock expires rastrillo's auth_links rows |
| 240 | // too. In production the caller passes the store's own clock, so it is |
| 241 | // simply time.Now. |
| 242 | func (a *app) sweepOnce(ctx context.Context, now time.Time) { |
| 243 | if n, err := a.store.sweep(ctx, now); err != nil { |
| 244 | a.logger.Error("calls: sweep pending joins", "err", err) |
| 245 | } else if n > 0 { |
| 246 | a.logger.Info("calls: swept pending joins", "deleted", n) |
| 247 | } |
| 248 | if err := a.auth.Sweep(now); err != nil { |
| 249 | a.logger.Error("calls: sweep auth links and sessions", "err", err) |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | // sweepHourly is what the ticker runs: store.sweep is bounded to |
| 254 | // sweepLimit rows a call (review C2-3), so a backlog bigger than that |
| 255 | // needs more than one call to actually clear it — the loop the boot |
| 256 | // sweep and SweepForTest deliberately do not do, because a deadline- |
| 257 | // bound or test-driven caller wants one page of work, not "until done". |
| 258 | func (a *app) sweepHourly(ctx context.Context) { |
| 259 | if n := a.overBudget.Swap(0); n > 0 { |
| 260 | a.logger.Warn("calls: outsider sign-in attempts over budget", "count", n) |
| 261 | } |
| 262 | now := a.store.now() |
| 263 | for { |
| 264 | n, err := a.store.sweep(ctx, now) |
| 265 | if err != nil { |
| 266 | a.logger.Error("calls: sweep pending joins", "err", err) |
| 267 | break |
| 268 | } |
| 269 | if n > 0 { |
| 270 | a.logger.Info("calls: swept pending joins", "deleted", n) |
| 271 | } |
| 272 | if n < sweepLimit { |
| 273 | break |
| 274 | } |
| 275 | } |
| 276 | if err := a.auth.Sweep(now); err != nil { |
| 277 | a.logger.Error("calls: sweep auth links and sessions", "err", err) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | // sweepForever is started by App and not by NewForTest: a test builds an |
| 282 | // app per case, and an hourly ticker per case is a goroutine per case |
| 283 | // that outlives it holding the store and the database open. It exits on |
| 284 | // ctx.Done, so App's caller can stop it by cancelling the context it |
| 285 | // passed in. |
| 286 | func (a *app) sweepForever(ctx context.Context) { |
| 287 | t := time.NewTicker(sweepEvery) |
| 288 | defer t.Stop() |
| 289 | for { |
| 290 | select { |
| 291 | case <-ctx.Done(): |
| 292 | return |
| 293 | case <-t.C: |
| 294 | a.sweepHourly(ctx) |
| 295 | } |
| 296 | } |
| 297 | } |
| 298 | |
| 299 | // tickEvery is the hub's housekeeping interval (spec §2.4): the quiet |
| 300 | // case, for participations nothing else is touching. |
| 301 | const tickEvery = 15 * time.Second |
| 302 | |
| 303 | // tickForever is started by App and not by NewForTest, for the reason |
| 304 | // sweepForever is: a ticker per test app is a goroutine per test that |
| 305 | // outlives it. Tests drive it with TestHandle.TickForTest. |
| 306 | func (a *app) tickForever(ctx context.Context) { a.tickLoop(ctx, tickEvery) } |
| 307 | |
| 308 | // tickLoop is the scheduling path, with the interval as a parameter so |
| 309 | // the one property that matters can be gated in milliseconds rather than |
| 310 | // in fifteen-second slots: the deadline each pass is given is the |
| 311 | // interval between passes, so a pass can never overrun the slot the next |
| 312 | // one needs. `every` goes to the ticker and to the pass, in one place, |
| 313 | // and there is no second constant to drift from the first. |
| 314 | // |
| 315 | // tickPass is nil everywhere but in that test, which substitutes a |
| 316 | // recorder for it; the alternative — asserting a constant equals itself |
| 317 | // — left the relationship unprotected (Codex S2-14). |
| 318 | func (a *app) tickLoop(ctx context.Context, every time.Duration) { |
| 319 | pass := a.tickPass |
| 320 | if pass == nil { |
| 321 | pass = a.tickOnce |
| 322 | } |
| 323 | t := time.NewTicker(every) |
| 324 | defer t.Stop() |
| 325 | for { |
| 326 | select { |
| 327 | case <-ctx.Done(): |
| 328 | return |
| 329 | case <-t.C: |
| 330 | pass(ctx, every) |
| 331 | } |
| 332 | } |
| 333 | } |
| 334 | |
| 335 | // tickOnce is one housekeeping pass under a deadline of its own (review |
| 336 | // W-12). The read phase does one standing lookup per joined |
| 337 | // participation, serially, with the hub lock down; given only the |
| 338 | // process context, a driver that stops answering would stall §2.4's |
| 339 | // quiet case for as long as it stayed stuck, and a removed member would |
| 340 | // then be dropped only by the next thing they did. A pass that runs |
| 341 | // past the interval has already lost its slot, so `within` is that |
| 342 | // interval: the reads that did answer are applied, the ones that did |
| 343 | // not wait for the next pass, which is what Tick already does with a |
| 344 | // failed lookup. |
| 345 | func (a *app) tickOnce(ctx context.Context, within time.Duration) { |
| 346 | ctx, cancel := context.WithTimeout(ctx, within) |
| 347 | defer cancel() |
| 348 | a.hub.Tick(ctx) |
| 349 | } |
| 350 | |
| 351 | // build wires the whole app: refusals, schema, router, static files. It |
| 352 | // returns the *app alongside the mux so tests can reach into it (see |
| 353 | // NewForTest); production code uses App, which discards the *app. |
| 354 | // |
| 355 | // ctx bounds the sweep at boot, and its cancellation is what tells live |
| 356 | // streams the process is stopping (a.done). The hourly ticker is not |
| 357 | // started here: App starts it, NewForTest does not, so a test suite does |
| 358 | // not leak a goroutine per app it builds. |
| 359 | func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, error) { |
| 360 | rc, err := resolve(cfg) |
| 361 | if err != nil { |
| 362 | return nil, nil, err |
| 363 | } |
| 364 | sealer, err := seal.New(rc.dek) |
| 365 | if err != nil { |
| 366 | return nil, nil, err |
| 367 | } |
| 368 | // rastrillo/db configures GORM's logger without ParameterizedQueries, |
| 369 | // so a slow or failing query is logged with its values interpolated |
| 370 | // into the SQL. ensurePerson looks a person up by address, so that |
| 371 | // one log line would carry an outsider's plaintext address (review |
| 372 | // C1-2). Same thresholds as db.Open, placeholders instead of values. |
| 373 | d.G.Logger = gormlogger.New(gormWriter{rc.Logger}, gormlogger.Config{ |
| 374 | SlowThreshold: 200 * time.Millisecond, |
| 375 | LogLevel: gormlogger.Warn, |
| 376 | IgnoreRecordNotFoundError: true, |
| 377 | ParameterizedQueries: true, |
| 378 | }) |
| 379 | |
| 380 | migrated, err := migrate.Apply(context.Background(), d, BootSchema) |
| 381 | if err != nil { |
| 382 | return nil, nil, err |
| 383 | } |
| 384 | rc.Logger.Info("migrate", "applied", migrated.Applied, |
| 385 | "skipped", migrated.Skipped, "adopted", migrated.Adopted) |
| 386 | |
| 387 | st := &store{g: d.G, sealer: sealer, clk: newClock(time.Now), logger: rc.Logger} |
| 388 | a := &app{ |
| 389 | db: d, cfg: rc, sealer: sealer, store: st, logger: rc.Logger, |
| 390 | linkBudget: newBudget(5, 15*time.Minute), |
| 391 | outsiderBudget: newBudget(10, 15*time.Minute), |
| 392 | } |
| 393 | salt := make([]byte, 16) |
| 394 | if _, err := rand.Read(salt); err != nil { |
| 395 | return nil, nil, err |
| 396 | } |
| 397 | a.hubSalt = hex.EncodeToString(salt) |
| 398 | a.done = ctx.Done() |
| 399 | a.heartbeat.Store(int64(defaultHeartbeat)) |
| 400 | a.ttl.Store(int64(defaultStreamTTL)) |
| 401 | a.relay.Store(rc.relay) |
| 402 | if len(rc.relay) > 0 { |
| 403 | rc.Logger.Info("calls: relay configured", "regions", len(rc.relay)) |
| 404 | } |
| 405 | // The hub reads standing back through the app, and never under its |
| 406 | // own lock: hubStanding is a database read. |
| 407 | // The APNs sender, where a key was configured. Built before the hub |
| 408 | // so that a.knocked has something to send with the first time it is |
| 409 | // called, which on a busy instance can be immediately. |
| 410 | if rc.apns != nil { |
| 411 | client, err := apns.New(*rc.apns) |
| 412 | if err != nil { |
| 413 | return nil, nil, fmt.Errorf("calls: Apple push: %w", err) |
| 414 | } |
| 415 | a.push = client |
| 416 | rc.Logger.Info("calls: Apple push configured", "sandbox", rc.APNsSandbox) |
| 417 | } |
| 418 | // The hub reads standing back through the app, and never under its |
| 419 | // own lock: hubStanding is a database read. Knocked is the same |
| 420 | // arrangement for the other direction. |
| 421 | a.hub = hub.New(hub.Config{ |
| 422 | Now: st.now, Logger: rc.Logger, StandingFor: a.hubStanding, Knocked: a.knocked, |
| 423 | WaitingRoom: func(ctx context.Context, sqid string) (bool, error) { |
| 424 | m, _, err := st.meetingBySqid(ctx, sqid) |
| 425 | return m.WaitingRoom, err |
| 426 | }, |
| 427 | }) |
| 428 | a.signalLimiter = newRateLimiter(30, 60, maxSignalBuckets) |
| 429 | a.apiLimiter = newRateLimiter(apiRatePerSecond, apiBurst, maxAgentBuckets) |
| 430 | a.qualityLimiter = newRateLimiterEvery(10*time.Second, 6, maxReportBuckets) |
| 431 | a.problemLimiter = newRateLimiterEvery(time.Minute, 3, maxReportBuckets) |
| 432 | a.reportLog = rc.ReportLogger |
| 433 | if !rc.qualityReports { |
| 434 | rc.Logger.Info("calls: quality reports off (CALLS_QUALITY_REPORTS=off)") |
| 435 | } |
| 436 | |
| 437 | writer, err := d.G.DB() |
| 438 | if err != nil { |
| 439 | return nil, nil, err |
| 440 | } |
| 441 | // Sign-in links go out after the response, from a worker that asks |
| 442 | // authorize first (signin_mail.go). It is NOT closed when ctx ends: |
| 443 | // ctx ends at SIGTERM, while Server.Shutdown is still letting |
| 444 | // accepted requests finish, and a request that reaches the queue |
| 445 | // after that must still have its link sent (Astra on #25, round 2). |
| 446 | // Whoever built the app closes it once no request can arrive: |
| 447 | // Runtime.Close (main, after rastrillo.Serve returns) or the |
| 448 | // TestHandle's stop. Only a failed build closes it here. |
| 449 | a.signinMail = newMailQueue(a.cfg.mailer, a.authorize, mayMail, a.logger) |
| 450 | if w, ok := cfg.Mailer.(queueAttacher); ok { |
| 451 | w.AttachSigninQueue(a.signinMail.wait) |
| 452 | } |
| 453 | authPlugin, err := a.newAuthPlugin(writer) |
| 454 | if err != nil { |
| 455 | a.signinMail.close() |
| 456 | return nil, nil, err |
| 457 | } |
| 458 | a.auth = authPlugin |
| 459 | if err := a.wireHome(); err != nil { |
| 460 | a.signinMail.close() |
| 461 | return nil, nil, err |
| 462 | } |
| 463 | |
| 464 | if err := a.wireNativeAuth(); err != nil { |
| 465 | a.signinMail.close() |
| 466 | return nil, nil, err |
| 467 | } |
| 468 | |
| 469 | // Bounded, not open-ended: a boot sweep runs on the caller's ctx but |
| 470 | // must still finish, so an app never fails to start (or hangs a |
| 471 | // test) because of a stuck sweep (review C2-3). |
| 472 | bootCtx, cancel := context.WithTimeout(ctx, 30*time.Second) |
| 473 | a.sweepOnce(bootCtx, a.store.now()) |
| 474 | cancel() |
| 475 | |
| 476 | r := chi.NewRouter() |
| 477 | // Every panic inside the router is the 500 page with a reference and |
| 478 | // the viewer's own chrome (errors.go); an unknown route is the 404 |
| 479 | // page, not chi's plain text (shared chrome spec §3.3). |
| 480 | r.Use(a.recoverToErrorPage) |
| 481 | r.NotFound(a.notFound) |
| 482 | r.MethodNotAllowed(a.methodNotAllowed(r)) |
| 483 | |
| 484 | // The drive surface (spec §8.2) is mounted at the router root, |
| 485 | // outside csrf.Protect and outside the session middlewares: an |
| 486 | // agent is not a browser and sends none of the headers |
| 487 | // csrf.SameOrigin needs (a POST carrying none of them is refused), |
| 488 | // while RequireSession would redirect its bearer to the sign-in |
| 489 | // page. apiRoutes says why that is safe — this surface honours no |
| 490 | // cookie, so there is no ambient credential for CSRF to protect. |
| 491 | r.Group(a.apiRoutes) |
| 492 | r.Post("/native/request", a.nativeRequest) |
| 493 | r.Post("/native/exchange", a.nativeExchange) |
| 494 | |
| 495 | r.Group(func(r chi.Router) { |
| 496 | if a.home == nil { |
| 497 | r.Use(csrf.Protect(rc.Origin)) |
| 498 | } else { |
| 499 | r.Use(a.home.routes.Middleware) |
| 500 | } |
| 501 | r.Use(a.nativeResume) |
| 502 | r.Get("/native/start", a.nativeStart) |
| 503 | r.Get("/help/open-source", a.credits) |
| 504 | // Appearance is inside the CSRF check and outside the session |
| 505 | // group: a session that expired mid-page must not turn the click |
| 506 | // into a sign-in loop (spec §8.3). |
| 507 | r.Post("/appearance", chrome.HandleAppearance) |
| 508 | r.Get(signinPath, a.signin) |
| 509 | // The handlers rastrillo/auth expects an app to mount. Verify |
| 510 | // answers GET with the confirm door (renderConfirm), which spends |
| 511 | // nothing, and POST with the redemption, so it is mounted for |
| 512 | // both methods in both modes, wrapped alike: without the POST the |
| 513 | // door's Sign in button gets a 405 and nobody signs in. The path |
| 514 | // is baked into every emailed link. A verified address authorize |
| 515 | // refuses is answered by auth's Refused hook (refused), inside |
| 516 | // Callback and Verify. |
| 517 | if a.home == nil { |
| 518 | r.Post(signinPath, a.newBegin(authPlugin.Begin)) |
| 519 | r.Get("/auth/callback", authPlugin.Callback) |
| 520 | r.Get("/auth/verify", authPlugin.Verify) |
| 521 | r.Post("/auth/verify", authPlugin.Verify) |
| 522 | r.Post("/signout", signedOutAfter(authPlugin.Signout)) |
| 523 | r.Get("/signed-out", a.signedOut) |
| 524 | } else { |
| 525 | r.Post(signinPath, a.home.routes.LocalGuard(a.newBegin(authPlugin.Begin))) |
| 526 | r.Get("/auth/callback", a.home.routes.LocalGuard(authPlugin.Callback)) |
| 527 | r.Get("/auth/verify", a.home.routes.LocalGuard(authPlugin.Verify)) |
| 528 | r.Post("/auth/verify", a.home.routes.LocalGuard(authPlugin.Verify)) |
| 529 | r.Post("/signout", a.home.routes.Signout) |
| 530 | r.Get("/signin/local", a.localSignin) |
| 531 | r.Get("/signed-out", a.home.routes.SignedOut) |
| 532 | r.Get("/auth/start", a.home.bridge.Start) |
| 533 | r.Post("/auth/assertion", a.home.bridge.Callback) |
| 534 | r.Get("/auth/finish", a.home.bridge.Finish) |
| 535 | r.Get("/auth/local/complete", a.home.routes.LocalComplete) |
| 536 | } |
| 537 | r.Get("/j/{token}", a.followLink) |
| 538 | r.Post("/j/{token}/guest", a.joinGuest) |
| 539 | |
| 540 | // A verified or guest session, with a persons row. Guest sessions are |
| 541 | // restricted to participation and native authorization routes. |
| 542 | r.Group(func(r chi.Router) { |
| 543 | r.Use(a.requireParticipant) |
| 544 | r.Get("/", a.index) |
| 545 | r.Get("/native/complete", a.nativeComplete) |
| 546 | r.Get("/native/authorize", a.nativeAuthorize) |
| 547 | r.Post("/native/authorize", a.nativeApprove) |
| 548 | r.Get("/continue", a.continueSignedIn) |
| 549 | r.Get("/welcome", a.welcome) |
| 550 | r.Get("/m/{sqid}/join", a.holding) |
| 551 | r.Get("/m/{sqid}/app", a.appOffer) |
| 552 | r.Get("/native/offer", a.offerPage) |
| 553 | r.Get("/m/{sqid}", a.room) |
| 554 | r.Get("/m/{sqid}/info", a.roomInfo) |
| 555 | r.Get("/m/{sqid}/remove/{peer}", a.confirmRemove) |
| 556 | r.Post("/m/{sqid}/remove/{peer}", a.remove) |
| 557 | r.Get("/m/{sqid}/takeover", a.confirmTakeover) |
| 558 | r.Post("/m/{sqid}/takeover", a.takeover) |
| 559 | r.Group(func(r chi.Router) { |
| 560 | r.Use(a.requireMember) |
| 561 | if a.home != nil { |
| 562 | r.Get("/account/home", a.home.routes.Account) |
| 563 | r.Post("/account/home", a.home.routes.LinkStart) |
| 564 | r.Get("/account/home/confirm", a.home.bridge.ConfirmPage) |
| 565 | r.Post("/account/home/confirm", a.home.routes.ConfirmPost) |
| 566 | } |
| 567 | r.Post("/meetings/start", a.startCall) |
| 568 | r.Post("/native/meetings/start", a.nativeStartMeeting) |
| 569 | r.Post("/native/devices", a.registerDevice) |
| 570 | r.Post("/meetings", a.createMeeting) |
| 571 | r.Post("/m/{sqid}/share", a.shareMeeting) |
| 572 | r.Get("/meetings/{sqid}/links/replace", a.confirmReplaceLink) |
| 573 | r.Post("/meetings/{sqid}/links", a.replaceLink) |
| 574 | r.Get("/meetings/{sqid}/end", a.confirmEnd) |
| 575 | r.Post("/meetings/{sqid}/end", a.endMeeting) |
| 576 | r.Get("/settings", a.settings) |
| 577 | r.Post("/settings/defaults", a.saveDefaults) |
| 578 | r.Get("/meetings/{sqid}/settings", a.meetingSettings) |
| 579 | r.Post("/meetings/{sqid}/settings", a.saveSettings) |
| 580 | r.Post("/settings/agents", a.createAgentHandler) |
| 581 | r.Get("/settings/agents/{id}/revoke", a.confirmRevokeAgent) |
| 582 | r.Post("/settings/agents/{id}/revoke", a.revokeAgentHandler) |
| 583 | }) |
| 584 | r.Get("/m/{sqid}/stream", a.stream) |
| 585 | r.Post("/m/{sqid}/signal", a.signal) |
| 586 | r.Get("/m/{sqid}/ice", a.ice) |
| 587 | r.Post("/m/{sqid}/quality", a.quality) |
| 588 | r.Post("/m/{sqid}/problem", a.problem) |
| 589 | }) |
| 590 | }) |
| 591 | |
| 592 | mux := http.NewServeMux() |
| 593 | mux.HandleFunc("GET /static/credits.css", serveCreditsCSS) |
| 594 | mux.Handle("GET /static/", assets.Handler()) |
| 595 | // The sky behind the sign-in door: estilo's closed set of public |
| 596 | // SVGs, sessionless, more specific than /static/ above. |
| 597 | mux.Handle("GET "+sky.Prefix, sky.Handler()) |
| 598 | // Onest, Calma's typeface, served by estilo under its versioned prefix. |
| 599 | mux.Handle("GET "+theme.FontPrefix, theme.FontHandler()) |
| 600 | mux.Handle("/", r) |
| 601 | return mux, a, nil |
| 602 | } |
| 603 | |
| 604 | // Runtime is what App hands main beyond the mux: the platform's |
| 605 | // scheduled-wake function (review C2-2), the error page, and the |
| 606 | // shutdown drain. It is a struct rather than a bare func so a later |
| 607 | // addition does not change App's signature again. |
| 608 | type Runtime struct { |
| 609 | // NextDue answers rastrillo.Options.NextDue: the next moment this |
| 610 | // app has work, or the zero time when nothing is scheduled. |
| 611 | NextDue func() time.Time |
| 612 | // ErrorPage answers rastrillo.Options.ErrorPage: the app's own error |
| 613 | // page for a panic outside the router. |
| 614 | ErrorPage func(w http.ResponseWriter, r *http.Request, status int, ref string) |
| 615 | // Close stops the sign-in mail queue taking links and drains it, |
| 616 | // within its bounded deadline, returning when that is done. The |
| 617 | // queue is NOT closed when App's ctx ends: that happens at SIGTERM, |
| 618 | // while Server.Shutdown still lets accepted requests finish, and |
| 619 | // their links must go out. So main calls Close after rastrillo.Serve |
| 620 | // returns (no request can arrive any more) and before the database |
| 621 | // closes; SMTP runs after the response, so Shutdown does not wait |
| 622 | // for it. Every caller of App must call Close, or the queue's worker |
| 623 | // outlives the app. A second call joins the first. |
| 624 | Close func() |
| 625 | } |
| 626 | |
| 627 | // nextDue is Runtime.NextDue's implementation: the earliest of the next |
| 628 | // nextExpiry (across pending_joins, auth_links and sessions) and one |
| 629 | // hour from now — the hourly sweep's own interval, so the platform |
| 630 | // never lets the app sleep past a sweep it would otherwise have run — |
| 631 | // or the zero time when nothing at all is outstanding, meaning there is |
| 632 | // nothing to wake early for. |
| 633 | // |
| 634 | // A read failure does NOT answer the zero time: zero means "nothing |
| 635 | // scheduled" to the platform (serve.go's contract), so returning it on |
| 636 | // a transient error would cancel the wake entirely rather than merely |
| 637 | // delay it. An hour out is the same retry deadline the healthy path |
| 638 | // already uses for "nothing due sooner than the next sweep," so a |
| 639 | // flaky read gets one more try at the next poll instead of going quiet |
| 640 | // (Codex round 3, C3-1). |
| 641 | func (a *app) nextDue() time.Time { |
| 642 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 643 | defer cancel() |
| 644 | retry := a.store.now().Add(sweepEvery) |
| 645 | exp, err := a.store.nextExpiry(ctx) |
| 646 | if err != nil { |
| 647 | a.logger.Error("calls: next due: read next expiry", "err", err) |
| 648 | return retry |
| 649 | } |
| 650 | if exp.IsZero() { |
| 651 | return time.Time{} |
| 652 | } |
| 653 | if retry.Before(exp) { |
| 654 | return retry |
| 655 | } |
| 656 | return exp |
| 657 | } |
| 658 | |
| 659 | // App wires the whole app: refusals, schema, router, static files. It |
| 660 | // returns a *http.ServeMux because rastrillo.Options.Mux is typed that |
| 661 | // way; the chi router mounts inside it, and a *Runtime carrying the |
| 662 | // platform wake function main.go wires to Options.NextDue. It also |
| 663 | // starts the hourly sweep and the hub tick on ctx, which run until ctx |
| 664 | // is cancelled or the process exits. |
| 665 | // |
| 666 | // Cancel ctx on SIGTERM: live SSE streams end when it is cancelled, and |
| 667 | // http.Server.Shutdown waits for handlers without cancelling their |
| 668 | // contexts, so a stream that does not watch this holds the shutdown |
| 669 | // deadline open. |
| 670 | func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, error) { |
| 671 | mux, a, err := build(ctx, d, cfg) |
| 672 | if err != nil { |
| 673 | return nil, nil, err |
| 674 | } |
| 675 | go a.sweepForever(ctx) |
| 676 | go a.tickForever(ctx) |
| 677 | // ctx's end stops streams, the sweep and the tick; the sign-in mail |
| 678 | // queue keeps taking links until Runtime.Close (see build). |
| 679 | return mux, &Runtime{NextDue: a.nextDue, ErrorPage: a.ErrorPage, Close: a.signinMail.close}, nil |
| 680 | } |
| 681 | |
| 682 | // NewForTest builds the app and returns the handler and its test |
| 683 | // controls. No sweep ticker is started: see build. Tests only; |
| 684 | // production uses App. |
| 685 | func NewForTest(d *db.DB, cfg Config) (*http.ServeMux, *TestHandle, error) { |
| 686 | ctx, cancel := context.WithCancel(context.Background()) |
| 687 | mux, a, err := build(ctx, d, cfg) |
| 688 | if err != nil { |
| 689 | cancel() |
| 690 | return nil, nil, err |
| 691 | } |
| 692 | // A test's stop (ShutdownForTest) ends the app outright, mail queue |
| 693 | // included: no server is draining requests in front of it. |
| 694 | go func() { <-ctx.Done(); a.signinMail.close() }() |
| 695 | return mux, &TestHandle{a: a, stop: cancel}, nil |
| 696 | } |
| 697 | |
amadan