diff --git a/AGENTS.md b/AGENTS.md index cf2ff2519..9a2189e43 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,7 +33,7 @@ All `265` operations across the ~50-service per-SDK layer are generated. Hand-wr |---------|----------| | HTTP helpers, pagination, hooks | `typescript/src/services/base.ts`, `ruby/lib/basecamp/generated/services/base_service.rb`, `swift/Sources/Basecamp/Services/BaseService.swift`, `kotlin/sdk/src/commonMain/kotlin/com/basecamp/sdk/services/BaseService.kt`, `python/src/basecamp/generated/services/_base.py`, `rust/basecamp-sdk/src/http/` | | OAuth flows (not in OpenAPI spec) | `typescript/src/services/authorization.ts`, `ruby/lib/basecamp/services/authorization_service.rb`, `kotlin/sdk/src/commonMain/kotlin/com/basecamp/sdk/oauth/`, `python/src/basecamp/services/authorization.py`, `rust/basecamp-sdk/src/oauth/` (no Swift equivalent) | -| Event Feed connector: the long-lived push+poll transport, hand-written behind its TicketMinter/PollSource seams (SPEC.md §23) | `go/pkg/basecamp/eventfeed/` (Go reference — foundations, the run loop, and the tier-2 conformance driver; the Layer-1 seam adapters over the generated operations and the other SDKs are still pending) | +| Event Feed connector: the long-lived push+poll transport, hand-written behind its TicketMinter/PollSource seams (SPEC.md §23) | `go/pkg/basecamp/eventfeed/` (Go reference — foundations, the run loop, the tier-2 conformance driver, and the Layer-1 seam adapters over the generated operations, `NewLive`; the other SDKs are still pending) | | Merge-safe Todos composites (update/edit over generated get+replace; SPEC.md §18) | `typescript/src/services/todos-extensions.ts`, `ruby/lib/basecamp/services/todos_extensions.rb`, `swift/Sources/Basecamp/TodosServiceExtensions.swift`, `kotlin/sdk/src/commonMain/kotlin/com/basecamp/sdk/services/TodosService.kt`, `python/src/basecamp/services/todos.py`, `rust/basecamp-sdk/src/services/todos.rs` | | Merge-safe Cards composite (update over generated get+updateVerbatim; SPEC.md §18) | `typescript/src/services/cards-extensions.ts`, `ruby/lib/basecamp/services/cards_extensions.rb`, `swift/Sources/Basecamp/CardsServiceExtensions.swift`, `kotlin/sdk/src/commonMain/kotlin/com/basecamp/sdk/services/CardsService.kt`, `python/src/basecamp/services/cards.py`, `rust/basecamp-sdk/src/services/cards.rs` | | Merge-safe Todolists composites (update/edit over generated get+replace; SPEC.md §18) | `typescript/src/services/todolists-extensions.ts`, `ruby/lib/basecamp/services/todolists_extensions.rb`, `swift/Sources/Basecamp/TodolistsServiceExtensions.swift`, `kotlin/sdk/src/commonMain/kotlin/com/basecamp/sdk/services/TodolistsService.kt`, `python/src/basecamp/services/todolists.py`, `rust/basecamp-sdk/src/services/todolists.rs` | diff --git a/SPEC.md b/SPEC.md index 1fb208f48..3a15b4125 100644 --- a/SPEC.md +++ b/SPEC.md @@ -4241,10 +4241,16 @@ END -- refresh and retry budget) → unauthorized; a 3xx whose Location fails the per-hop -- same-origin/no-downgrade validation (auto-follow is disabled — Continuation and -- Resume URL Validation) → redirect_refused, carrying the refused Location redacted to --- its origin → Terminal(`invalid_continuation`), NEVER unrecoverable; anything else +-- its origin — or, in its place, the fixed token `unparsable` when the Location yielded +-- no complete origin (§9) and `unrecorded` when the hop was refused but the adapter +-- could not attribute the origin to the call; the refusal and its zero egress never +-- depend on that attribution, and the two tokens are never interchanged, since one +-- reports what the server sent and the other what the adapter failed to keep +-- → Terminal(`invalid_continuation`), NEVER unrecoverable; anything else -- non-retryable (404, 405, unexpected shapes) → unrecoverable, carrying the generated --- error verbatim. A same-origin Location may be followed inside the seam under the same --- per-hop rule (no error surfaces). +-- error verbatim. No Location is followed inside the seam, same-origin included: the API +-- never redirects a feed call, and a continuation is followed by re-issuing the +-- operation, never by a hop. INTERFACE CableTransport dial(ws_url, cancellation, max_frame_bytes) → CableConn @@ -4352,11 +4358,11 @@ not an operable feed state. stacks auto-follow redirects (Go strips `Authorization` on a cross-origin hop but still egresses), which would falsify the zero-foreign-egress guarantee the moment a validated same-origin URL answers 3xx with a foreign `Location`. The Layer-1 adapter therefore -**disables automatic redirect-following for `PollEvents`** (or per-hop validates every -resolved `Location` under §8's hop-anchored rule): a 3xx from a validated URL yields its -`Location` to the same same-origin + no-downgrade validation — cross-origin or downgraded -→ Terminal(`invalid_continuation`) with zero egress to the foreign origin; same-origin → -it may be followed, each hop under the same rule. +**answers every 3xx to a feed operation itself and follows none**: a 3xx from a validated +URL yields its `Location` reduced to its origin, with the hop refused — foreign, +downgraded and same-origin alike, since the API never redirects a feed call and a +continuation is followed by re-issuing the operation — → Terminal(`invalid_continuation`) +with zero egress to the foreign origin. The mint's cable `url` is deliberately **not** under this rule: it is server-directed cable topology, cross-host by design, dialed verbatim with its own credential (the @@ -4383,9 +4389,13 @@ decides whether to follow one, which makes the foreign origin unreachable by construction of the harness — a harness that asserted no request reached it would be asserting something about itself. That obligation belongs to the Layer-1 seam adapter's own 302 test, where a real generated `PollEvents` call meets a real redirect -against an adapter with automatic redirect-following disabled. -`conformance/event-feed/README.md`'s row-15 note records it as a pending obligation -rather than a proof the repository contains; the adapters are tracked in #819. +against an adapter whose client refuses the hop: the Go adapters (`eventfeed.NewLive`) +compose a guard over their client's transport that answers every 3xx at the wire — +the `Location` reduced to its origin for the seam and stripped, with the body, before +the HTTP stack, the operation hooks or any log sees it; no hop is followed, same-origin +included, since the API never redirects a feed call — and their 302 test proves zero +egress against a sentinel listener behind the foreign `Location` +(`conformance/event-feed/README.md`, row 15). ### Clock, Timers, and Virtual Time `[conformance]` diff --git a/conformance/event-feed/README.md b/conformance/event-feed/README.md index d399b3fef..054baee2d 100644 --- a/conformance/event-feed/README.md +++ b/conformance/event-feed/README.md @@ -320,7 +320,7 @@ revoked-mint threshold). | 27 | `27-hostile-resume-cross-origin.json` | accepted 410 with a cross-origin `resume` → Terminal(`invalid_continuation`), zero foreign requests | | 28 | `28-checkpoint-load-failure.json` | store load Failed → Terminal(`checkpoint_load`) with ZERO wire attempts; distinct from Missing (which proceeds to a present entry) | | 29 | `29-checkpoint-save-failure-continues.json` | save Failed → feed continues and a SUBSEQUENT save is attempted (exact store-call script: no save circuit breaker) | -| 30 | `30-continuation-redirect-cross-origin.json` | validated same-origin `next` answering 302 + cross-origin Location → Terminal(`invalid_continuation`); zero foreign egress holds by construction of the seam here, and proving it against a real redirect is ASSIGNED to Layer 1, whose adapters are still pending, tracked in #819 — see the row-15 note | +| 30 | `30-continuation-redirect-cross-origin.json` | validated same-origin `next` answering 302 + cross-origin Location → Terminal(`invalid_continuation`); zero foreign egress holds by construction of the seam here, and proving it against a real redirect is ASSIGNED to Layer 1, whose Go adapters' 302 test proves it against a sentinel listener — see the row-15 note | | 31 | `31-post-snapshot-straggler-below-served-id.json` | post-snapshot straggler with an id BELOW the entry page's served id delivered live; the re-push of that served id still suppressed | | 32 | `32-inbox-lane-dedupes-by-addressing-id.json` | the inbox lane: `inbox: true` + `reasons` on the subscription; `{items, position, next}` pages; two items over one event both deliver; an addressing id re-served by the repair poll after its live delivery is suppressed; delivered ids are addressing ids | | 33 | `33-inbox-gap-410-retention-resume.json` | the inbox's 410 is the retention window: accepted, the resume URL (`since=0`, the earliest retained item) is followed as a position-resume entry and its page saves on acceptance | @@ -415,8 +415,8 @@ when every line is done: Fifteen of the sixteen mutations are shown red against at least one fixture in the reference implementation PR's body before they count. Row 15 is the -recorded exception — not killed at tier 2, pending the Layer-1 adapters #819 -tracks — and the note below is its account. +recorded exception — not killed at tier 2; killed by the Layer-1 adapters' own 302 +test — and the note below is its account. **One row is an exception, and it is the reason this heading is worth reading twice.** Row 15's mutation is **not killed at tier 2 at all** — it lives below @@ -442,7 +442,7 @@ exactly the class of claim this family exists to check. | 12 | `bypass-configured-handler` (handler registered but skipped; default-terminal applied) | 24, 25 (via `handlerInvocations` exact-set) | | 13 | `follow-cross-origin-continuation` (skips §8 validation, polls the hostile URL) | 26, 27 | | 14 | `collapse-load-error-to-missing` | 28 | -| 15 | `follow-cross-origin-redirect` (follows a 302 to a foreign Location) | **not killed at tier 2** — below the poll seam; assigned to Layer 1, whose adapters are still pending, tracked in #819. Fixture 30 pins a different fault class above the seam. See the note under this table. | +| 15 | `follow-cross-origin-redirect` (follows a 302 to a foreign Location) | **not killed at tier 2** — below the poll seam; killed at Layer 1 by the Go adapters' `TestLivePolls_RefusesACrossOriginRedirectWithZeroEgress` (a real generated `PollEvents` call meets a real 302, the hop is refused before any request, and a sentinel listener behind the foreign Location records zero hits). Fixture 30 pins a different fault class above the seam. See the note under this table. | | 16 | `discard-live-id-at-or-below-served-id` (streaming lane orders live ids against the highest poll-served id) | 31 — and 31 alone: verified to pass all of 01–30, because every other straggler either arrives with nothing yet served (20) or is buffered pre-cut (01, 12, 19) | **Row 15 is not killed at tier 2, and the reason is structural.** In tier 2 the @@ -476,11 +476,14 @@ That is withdrawn. No implementation met it, and meeting it would prove nothing: the foreign origin is unreachable **by construction of the harness**, because the harness is the seam, so a silent sentinel is a statement about the driver rather than about the connector. Zero egress to a foreign redirect -target is a Layer-1 property, and proving it is ASSIGNED to the Layer-1 seam -adapter's own 302 test, where a real generated `PollEvents` call will meet a -real redirect. Those adapters have not landed — `go/pkg/basecamp/eventfeed/doc.go` -lists them among the pieces still to come — so this is a recorded obligation, -not a proof the repository contains today. Tracked in #819. +target is a Layer-1 property, and proving it belongs to the Layer-1 seam +adapter's own 302 test, where a real generated `PollEvents` call meets a real +redirect. The Go adapters (`eventfeed.NewLive`, `go/pkg/basecamp/eventfeed/live.go`) +carry that test: a guard composed over their client's transport answers every 3xx +at the wire — no hop is followed, the `Location` is reduced to its origin and +stripped before anything else sees it — and a sentinel listener behind the +foreign `Location` records zero hits. The other SDKs' adapters owe the same +test when they land. Auto-continue-past-unhandled-gap needs no separate mutation — fixture 23's exact-set `finally` is its direct test. Fixture 29's exact store-call script is the diff --git a/conformance/event-feed/fixtures/30-continuation-redirect-cross-origin.json b/conformance/event-feed/fixtures/30-continuation-redirect-cross-origin.json index 317744fe0..a35312ce8 100644 --- a/conformance/event-feed/fixtures/30-continuation-redirect-cross-origin.json +++ b/conformance/event-feed/fixtures/30-continuation-redirect-cross-origin.json @@ -1,6 +1,6 @@ { "name": "30-continuation-redirect-cross-origin", -"description": "A VALIDATED same-origin `next` answers 302 with a cross-origin Location: the poll seam suppresses automatic redirect-following, and the foreign Location is Terminal(invalid_continuation). {{NEXT:1}} substitutes same-origin, so the pre-poll validation PASSES and the second poll seam call is made (contrast fixture 26, where no request reaches the URL at all) \u2014 the redirect answer is where the per-hop rule bites. What tier 2 verifies is the per-hop refusal DECISION \u2014 the driver classifies the scripted Location through the SHIPPED predicate (checkContinuation, via a test export), never an ad-hoc re-implementation \u2014 and the loop's response: the continuation terminal, the socket explicitly closed, and no further seam calls. That the adapter's HTTP client makes zero requests to the refused URL is below the poll seam and owned by Layer-1 adapter conformance \u2014 a recorded obligation, not an existing proof: those adapters are still pending (tracked in #819), so follow-cross-origin-redirect itself remains NOT killed at tier 2 (row 15). The Location host is literal, never substituted, and never served. Redaction is deliberately not claimed here: the driver reduces the Location to its origin best-effort (an unreducible Location carries none) before the connector runs, so no path or query text exists for a connector to over-echo; TestRedirectRefusalRendersNoServerValue owns that proof. See the row-15 note in README.md.", "config": { +"description": "A VALIDATED same-origin `next` answers 302 with a cross-origin Location: the poll seam suppresses automatic redirect-following, and the foreign Location is Terminal(invalid_continuation). {{NEXT:1}} substitutes same-origin, so the pre-poll validation PASSES and the second poll seam call is made (contrast fixture 26, where no request reaches the URL at all) \u2014 the redirect answer is where the per-hop rule bites. What tier 2 verifies is the per-hop refusal DECISION \u2014 the driver classifies the scripted Location through the SHIPPED predicate (checkContinuation, via a test export), never an ad-hoc re-implementation \u2014 and the loop's response: the continuation terminal, the socket explicitly closed, and no further seam calls. That the adapter's HTTP client makes zero requests to the refused URL is below the poll seam and owned by Layer-1 adapter conformance \u2014 the Go adapters' 302 test proves it against a sentinel listener \u2014 so follow-cross-origin-redirect itself remains NOT killed at tier 2 (row 15). The Location host is literal, never substituted, and never served. Redaction is deliberately not claimed here: the driver reduces the Location to its origin best-effort (an unreducible Location carries none) before the connector runs, so no path or query text exists for a connector to over-echo; TestRedirectRefusalRendersNoServerValue owns that proof. See the row-15 note in README.md.", "config": { "position": "{{POS:0}}" }, "steps": [ diff --git a/conformance/event-feed/schema.json b/conformance/event-feed/schema.json index c7aec55c4..a219f1219 100644 --- a/conformance/event-feed/schema.json +++ b/conformance/event-feed/schema.json @@ -1411,7 +1411,7 @@ "status", "headers" ], - "description": "Redirect on a continuation (fixture 30): the seam suppresses automatic following; a cross-origin/downgraded Location is Terminal(invalid_continuation). At tier 2 the poll lane is a SEAM, so the driver forms the redirect-refused verdict and the connector never sees a Location header: this pins the fault class above the seam (mishandling the verdict — retrying it or misclassifying it), not redirect-following itself, which lives below the seam, is unreachable at tier 2, and is ASSIGNED to the Layer-1 adapter's 302 test — a recorded obligation, not an existing proof: those adapters are still pending (tracked in #819). Redaction is not pinned here: the driver reduces the Location to its origin before the connector runs, so no path or query text reaches it. No sentinel-listener obligation is imposed on the harness: the foreign origin is unreachable by construction of the harness, so a silent sentinel would be a statement about the driver rather than about the connector. See the row-15 note in README.md.", + "description": "Redirect on a continuation (fixture 30): the seam suppresses automatic following; a cross-origin/downgraded Location is Terminal(invalid_continuation). At tier 2 the poll lane is a SEAM, so the driver forms the redirect-refused verdict and the connector never sees a Location header: this pins the fault class above the seam (mishandling the verdict — retrying it or misclassifying it), not redirect-following itself, which lives below the seam, is unreachable at tier 2, and is the Layer-1 adapter's 302 test (the Go adapters prove zero egress against a sentinel listener; the other SDKs' adapters owe the same). Redaction is not pinned here: the driver reduces the Location to its origin before the connector runs, so no path or query text reaches it. No sentinel-listener obligation is imposed on the harness: the foreign origin is unreachable by construction of the harness, so a silent sentinel would be a statement about the driver rather than about the connector. See the row-15 note in README.md.", "properties": { "status": { "const": 302 diff --git a/go/README.md b/go/README.md index 2e421e928..1bf8b6c1e 100644 --- a/go/README.md +++ b/go/README.md @@ -806,14 +806,16 @@ timer, and resumes after a disconnect. You consume it as one serial, deduplicate stream of events; the connector owns reconnection, backoff, staleness detection, and the durable position. -**Experimental: the Layer-1 seam adapters have not landed yet.** The connector performs -no HTTP API I/O of its own: every HTTP exchange reaches the wire through a seam backed by -a generated operation. Its one direct wire act is the Action Cable dial above — the -connector connects verbatim to the URL a generated `CreateStreamTicket` call returned, -which is the sanctioned non-HTTP wire act. The adapters that build those seams over the -generated `CreateStreamTicket` and `PollEvents` operations are still to come. Until they -do, a consumer must supply the `TicketMinter` and `PollSource` implementations itself, and -the exported surface may still change as they land. +**Experimental.** The connector performs no HTTP API I/O of its own: every HTTP +exchange reaches the wire through a seam backed by a generated operation, and its one +direct wire act is the Action Cable dial above — the connector connects verbatim to the +URL a generated `CreateStreamTicket` call returned, which is the sanctioned non-HTTP wire +act. `eventfeed.NewLive` binds those seams to the generated `CreateStreamTicket`, +`PollEvents` and `PollInbox` operations over a `basecamp.Client` it builds for you — with +a transport guard that answers every 3xx at the wire instead of following it, so an +authenticated poll never egresses off the API origin and no `Location` reaches a log — +and `Connect` builds the connector over them. A host may still +supply its own `TicketMinter` and `PollSource`; the exported surface may still change. ```go import ( @@ -822,21 +824,20 @@ import ( "fmt" "log" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" ) -// The two seams the host supplies until the Layer-1 adapters land. Each call is -// exactly one generated operation: -// -// MintStreamTicket(ctx) (eventfeed.StreamTicket, error) // CreateStreamTicket -// Poll(ctx, cursor, filters) (eventfeed.PollPage, error) // PollEvents -var minter eventfeed.TicketMinter -var polls eventfeed.PollSource +var tokenProvider basecamp.TokenProvider var agentPersonID int64 // the acting principal's own person id ctx := context.Background() -feed, err := eventfeed.New("https://3.basecampapi.com", "5951425", minter, polls, +live, err := eventfeed.NewLive(basecamp.DefaultConfig(), tokenProvider, "5951425", eventfeed.AccountLane) +if err != nil { + return err +} +feed, err := live.Connect( eventfeed.WithFilters(eventfeed.Filters{ Types: []string{"message.created"}, // The loop guard for an agent that acts on what it hears: its own @@ -890,9 +891,13 @@ attempts. `Events` is single-shot: consuming it twice yields one `ReasonUsage` error element. -The same connector consumes the principal's **inbox** — the low-noise lane of items that -addressed the agent (mentions, assignments, subscriptions, watches, pings, boosts) — with -`eventfeed.WithLane(eventfeed.InboxLane)` and a `PollSource` over the inbox endpoint. +A second binding consumes the principal's **inbox** — the low-noise lane of items that +addressed the agent (mentions, assignments, subscriptions, watches, pings, boosts): +`eventfeed.NewLive(cfg, tokenProvider, "5951425", eventfeed.InboxLane)` binds the seams +to `PollInbox`, and its `Connect` builds the inbox connector (the lane is the binding's; +a `WithLane` naming the other lane is refused, never silently overridden). A host that +supplies its own seams passes `eventfeed.WithLane(eventfeed.InboxLane)` to `New` with a +`PollSource` over the inbox endpoint. Every delivered event then carries `ev.Addressing` (the item id, the reason, when), and the connector deduplicates and positions by that item id, since one event can address you for several reasons. Filter it with `Filters{Reasons: ...}` (plus `Types` and diff --git a/go/pkg/basecamp/client.go b/go/pkg/basecamp/client.go index 6e444a38c..f37fd8a89 100644 --- a/go/pkg/basecamp/client.go +++ b/go/pkg/basecamp/client.go @@ -39,6 +39,10 @@ type Client struct { logger *slog.Logger httpOpts HTTPOptions hooks Hooks + // transportWrapper, when set, wraps the resolved transport before the + // logging layer (WithTransportWrapper). An interface, not a func: Client + // stays comparable. + transportWrapper TransportWrapper // Generated client (single shared instance, account passed per operation) genOnce sync.Once @@ -155,6 +159,25 @@ func WithHTTPClient(c *http.Client) ClientOption { } } +// TransportWrapper composes a RoundTripper over the transport a Client +// resolves — the one WithTransport supplied, or the default — so a caller can +// observe or rewrite every wire response without replacing the transport +// beneath it. The wrapper sits under the client's logging layer: what it +// returns is what hooks, logs and the redirect policy see. The SPEC §23 event +// feed connector installs one that answers every 3xx itself, because an +// authenticated poll must never egress to a foreign origin at all. +type TransportWrapper interface { + WrapTransport(inner http.RoundTripper) http.RoundTripper +} + +// WithTransportWrapper installs the TransportWrapper the client composes over +// its transport. A later option replaces an earlier one. +func WithTransportWrapper(w TransportWrapper) ClientOption { + return func(client *Client) { + client.transportWrapper = w + } +} + // WithUserAgent sets the User-Agent header. func WithUserAgent(ua string) ClientOption { return func(client *Client) { @@ -236,6 +259,10 @@ func NewClient(cfg *Config, tokenProvider TokenProvider, opts ...ClientOption) * transport = newDefaultTransport() } + if c.transportWrapper != nil { + transport = c.transportWrapper.WrapTransport(transport) + } + // Wrap transport with logging transport transport = &loggingTransport{inner: transport, client: c} diff --git a/go/pkg/basecamp/event_feed.go b/go/pkg/basecamp/event_feed.go index 832393d48..05677d97d 100644 --- a/go/pkg/basecamp/event_feed.go +++ b/go/pkg/basecamp/event_feed.go @@ -601,7 +601,10 @@ func splitInt64s(joined, name string) ([]int64, error) { for _, part := range parts { v, err := strconv.ParseInt(part, 10, 64) if err != nil { - return nil, ErrUsage(fmt.Sprintf("continuation URL %s filter %q is not an integer id", name, part)) + // The value is server-written text a continuation must never + // render (SPEC §23 "Continuation and Resume URL Validation"): + // the message names the filter, never the value. + return nil, ErrUsage(fmt.Sprintf("continuation URL %s filter carries a value that is not an integer id", name)) } values = append(values, v) } diff --git a/go/pkg/basecamp/eventfeed/doc.go b/go/pkg/basecamp/eventfeed/doc.go index 2e4d25dda..4c5f11ef7 100644 --- a/go/pkg/basecamp/eventfeed/doc.go +++ b/go/pkg/basecamp/eventfeed/doc.go @@ -20,17 +20,16 @@ // addressed items, where the delivered Event carries its Addressing and the // addressing id is the lane's identity for dedupe and positioning. // -// ONE piece is still to land, and it is what keeps the package from running -// against the live API out of the box: the Layer-1 adapters over the -// generated CreateStreamTicket, PollEvents and PollInbox operations that -// back the TicketMinter and PollSource seams. Until they exist the package ships no -// implementation of those two seams, and a host that wants the live feed -// supplies its own over the generated operations — a supported path, and -// the one the seam contracts are written for, not a workaround. Two -// obligations ride on whichever adapters back the seams rather than on -// anything here — zero egress to a foreign redirect target -// (conformance/event-feed/README.md's row-15 note), and the -// no-automatic-redirect-following rule §23 places on PollEvents. +// NewLive binds the seams to the generated CreateStreamTicket, PollEvents +// and PollInbox operations through the basecamp client: Live.Minter and +// Live.Polls are the TicketMinter and PollSource a live feed runs on, and +// Live.Connect builds the Connector over them. The client it constructs +// refuses cross-origin and downgraded redirects before any request is +// issued, which is how the two obligations the seam contracts place on +// whichever adapters back them — zero egress to a foreign redirect target +// and no automatic redirect-following on the poll lane — are met here. A +// host may still supply its own seams over the generated operations; that +// is the path the seam contracts are written for, not a workaround. // // # Seams-first architecture // diff --git a/go/pkg/basecamp/eventfeed/export_test.go b/go/pkg/basecamp/eventfeed/export_test.go index b5006ad9a..da2c87cf2 100644 --- a/go/pkg/basecamp/eventfeed/export_test.go +++ b/go/pkg/basecamp/eventfeed/export_test.go @@ -2,6 +2,7 @@ package eventfeed import ( "context" + "encoding/json" "errors" "time" ) @@ -148,6 +149,36 @@ func ExportSubscribeFrame(f Filters) []byte { return subscribeCommand(subscribeIdentifier(AccountLane, f)) } +// ExportMapPollErrorKind exposes the poll seam's error classification for an +// error that never reached the wire. +func ExportMapPollErrorKind(err error) PollErrorKind { + var pe *PollError + if errors.As(mapPollError(context.Background(), err, &refusedHop{}, AccountLane), &pe) { + return pe.Kind + } + return 0 +} + +// ExportMapMintErrorKind is ExportMapPollErrorKind for the mint seam. +func ExportMapMintErrorKind(err error) MintErrorKind { + var me *MintError + if errors.As(mapMintError(context.Background(), err, &refusedHop{}), &me) { + return me.Kind + } + return 0 +} + +// ExportDecodePushEvent decodes one event object the way the push lane does, +// so a test can hold the poll adapter to the same bytes. +func ExportDecodePushEvent(raw []byte) (Event, error) { + return decodeEventObject(json.RawMessage(raw), true) +} + +// ExportIsFeedOperationPath exposes the redirect guard's route match. +func ExportIsFeedOperationPath(basePath, path string) bool { + return isFeedOperationPath(basePath, path) +} + // ExportInboxSubscribeFrame is ExportSubscribeFrame for the inbox lane. func ExportInboxSubscribeFrame(f Filters) []byte { return subscribeCommand(subscribeIdentifier(InboxLane, f)) diff --git a/go/pkg/basecamp/eventfeed/live.go b/go/pkg/basecamp/eventfeed/live.go new file mode 100644 index 000000000..175fed7e1 --- /dev/null +++ b/go/pkg/basecamp/eventfeed/live.go @@ -0,0 +1,937 @@ +package eventfeed + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "path" + "slices" + "strconv" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" +) + +// The Layer-1 adapters: the TicketMinter and PollSource seams bound to the +// generated CreateStreamTicket, PollEvents and PollInbox operations through +// the basecamp.EventFeedService wrapper (SPEC.md §23 "Seam Contracts"). One +// seam call is one fully-governed generated call — the operation keeps its +// SPEC §7 retry budget, backoff and Retry-After inside the seam, and each +// adapter maps every §6/§7 outcome onto exactly one seam error kind. +// +// The adapters build their own basecamp.Client rather than borrowing the +// host's, for one reason: redirects. A followed `next` or 410 `resume` URL +// carries the caller's bearer, and the client's default policy follows a +// cross-origin hop with the Authorization header stripped — which still +// egresses to the foreign origin. §23 "Continuation and Resume URL +// Validation" requires zero foreign egress, so the feed's client composes a +// guard over the host's transport (basecamp.WithTransportWrapper) that +// answers every 3xx of a seam call at the wire: the Location is reduced to +// its origin for the seam and stripped — with the body — before net/http, +// the operation hooks or any log sees it, and the 3xx reaches the generated +// call as a status it classifies. No hop is ever followed, same-origin +// included: the API never redirects a feed call, and a continuation is +// followed by re-issuing the operation, not by a hop. The guard acts only on +// the seams' own calls — it recognizes them by their routes beneath the +// configured base path — so every other request through the same client +// keeps the client's own redirect handling: a host can use it for its own +// refetches (Live.Client), downloads included, whose first hop legitimately +// 302s to a signed URL. The host keeps every other option — its hooks, logger, +// transport, auth strategy — by passing them through. + +// Live is the connector's wire binding: the seams for one account on one +// lane, over the generated operations. +type Live struct { + client *basecamp.Client + svc *basecamp.EventFeedService + origin string + accountID string + lane Lane +} + +// NewLive builds the seams for accountID on lane over a basecamp.Client +// constructed from cfg, tokens and clientOpts — with the feed's redirect +// guard installed last, so it wins over any transport wrapper in clientOpts +// (the host's transport itself, from WithTransport, is what the guard +// composes over). The client's base URL is the connector's origin: the checkpoint key's and the +// same-origin reference every continuation is validated against. +func NewLive(cfg *basecamp.Config, tokens basecamp.TokenProvider, accountID string, lane Lane, clientOpts ...basecamp.ClientOption) (*Live, error) { + if cfg == nil { + return nil, usageError("a basecamp.Config is required") + } + if accountID == "" { + return nil, usageError("accountID must be non-empty") + } + for _, r := range accountID { + if r < '0' || r > '9' { + // ForAccount panics on anything but digits; a configuration + // value reaches this constructor as an error, never a panic. + return nil, usageError("accountID must be numeric") + } + } + if lane != AccountLane && lane != InboxLane { + return nil, usageError(fmt.Sprintf("unknown lane %s", lane)) + } + // The base URL is host configuration, and it reaches the request hooks + // and logs whole through every request's URL, so a value carrying + // userinfo is refused here — with a fixed message, since echoing it would + // be the leak — and the origin is canonicalized from what remains. + if u, err := url.Parse(cfg.BaseURL); err != nil || u.User != nil || !utf8.ValidString(cfg.BaseURL) { + // Invalid UTF-8 is refused raw, as New refuses it: canonicalization + // would rewrite every invalid byte to U+FFFD and let two different + // broken origins collapse into one checkpoint lineage. + return nil, usageError("the base URL must be valid UTF-8, parse, and carry no userinfo") + } + origin, err := CanonicalOrigin(cfg.BaseURL) + if err != nil { + return nil, usageError("the base URL does not name an origin") + } + if err := checkOriginScheme(origin); err != nil { + return nil, err + } + opts := make([]basecamp.ClientOption, 0, len(clientOpts)+1) + opts = append(opts, clientOpts...) + // The guard is anchored at the base URL's path, so that path must be + // the one the generated client resolves routes beneath: a dot segment + // or a doubled slash would resolve to a different prefix than the one + // recorded here, and the guard would miss the seams' own calls. + basePath := "/" + if u, _ := url.Parse(cfg.BaseURL); u != nil { + if u.RawQuery != "" || u.Fragment != "" || u.ForceQuery || u.RawFragment != "" { + return nil, usageError("the base URL must carry no query or fragment") + } + // The rules below read the DECODED path, and the client sends the + // escaped one: `/api%2f` decodes to `/api/`, passes both of them, + // and goes on the wire as a request whose path is `/api%2f/...` — + // where the guard, anchored at `/api/`, finds no account segment + // and lets the 3xx through to be followed, off the API origin, + // with the request's query. RawPath is non-empty exactly when the + // escaped path is not the canonical encoding of the decoded one, + // so refusing it here closes every escaped spelling rather than + // the one that was found. + if u.RawPath != "" { + return nil, usageError("the base URL path must be canonical: no percent-encoded segments") + } + if strings.Contains(u.Path, "//") { + return nil, usageError("the base URL path must be canonical: no dot segments, no doubled slashes") + } + if strings.Trim(u.Path, "/") != "" { + if clean := path.Clean(u.Path); clean != strings.TrimSuffix(u.Path, "/") { + return nil, usageError("the base URL path must be canonical: no dot segments, no doubled slashes") + } + basePath = "/" + strings.Trim(u.Path, "/") + "/" + } + } + opts = append(opts, basecamp.WithTransportWrapper(redirectGuardWrapper{basePath: basePath})) + client := basecamp.NewClient(cfg, tokens, opts...) + return &Live{ + client: client, + svc: client.ForAccount(accountID).EventFeed(), + origin: origin, + accountID: accountID, + lane: lane, + }, nil +} + +// Client is the basecamp.Client the seams call through. A host that refetches +// the resources the feed points at can use it for those calls too. +func (l *Live) Client() *basecamp.Client { return l.client } + +// Origin is the canonical API origin the client is bound to — pass it to New. +func (l *Live) Origin() string { return l.origin } + +// Minter is the TicketMinter seam over CreateStreamTicket. +func (l *Live) Minter() TicketMinter { return &liveMinter{svc: l.svc} } + +// Polls is the PollSource seam over PollEvents (AccountLane) or PollInbox +// (InboxLane). Each call is a fresh source with its own walk state — one +// per connector, so a walk's order is held across its pages and two +// connectors over one binding never read each other's. +func (l *Live) Polls() PollSource { return &livePolls{svc: l.svc, lane: l.lane} } + +// Connect builds the Connector over these seams: New with this binding's +// origin, account and lane, plus opts. The lane is the binding's — the seams +// are bound to one lane's operation — so a WithLane in opts naming another +// lane is a usage error, never a silent override in either direction. +func (l *Live) Connect(opts ...Option) (*Connector, error) { + all := make([]Option, 0, len(opts)+1) + all = append(all, WithLane(l.lane)) + all = append(all, opts...) + c, err := New(l.origin, l.accountID, l.Minter(), l.Polls(), all...) + if err != nil { + return nil, err + } + if c.cfg.lane != l.lane { + return nil, usageError(fmt.Sprintf("the lane is bound by NewLive (%s); build a second Live for the %s lane", l.lane, c.cfg.lane)) + } + return c, nil +} + +// refusedHop is the per-call record the redirect guard writes when it +// answers a 3xx: the refused Location reduced to its origin — data for the +// seam's redirect_refused kind, never rendered (§23: a hostile redirect can +// reflect the bearer into a host label). It travels on the call's context, +// the only per-call channel between a seam call and the transport under it, +// so the guard — which sees only the wire exchange — can hand the origin +// back to the call that owns it. The refusal itself does not depend on the +// record: the guard answers a 3xx off the request's route, so a record that +// does not survive costs the origin, never the egress guarantee. A host hook +// that returns a context unrelated to the one it was handed is what loses it +// — the same channel carries the client's own per-request state (the retry +// attempt, the download projection marker) and the caller's cancellation — +// and the seam then reports `unrecorded` rather than claiming the server +// sent a Location it could not parse. +type refusedHop struct { + mu sync.Mutex + refused bool + origin string +} + +type refusedHopKey struct{} + +func (h *refusedHop) record(origin string) { + h.mu.Lock() + defer h.mu.Unlock() + h.refused, h.origin = true, origin +} + +func (h *refusedHop) get() (string, bool) { + h.mu.Lock() + defer h.mu.Unlock() + return h.origin, h.refused +} + +// withRefusedHop attaches a fresh record to ctx for one seam call. +func withRefusedHop(ctx context.Context) (context.Context, *refusedHop) { + hop := &refusedHop{} + return context.WithValue(ctx, refusedHopKey{}, hop), hop +} + +// redirectGuardWrapper installs redirectGuard over the client's transport, +// anchored at the configured base URL's path ("/" when it has none). +type redirectGuardWrapper struct{ basePath string } + +func (w redirectGuardWrapper) WrapTransport(inner http.RoundTripper) http.RoundTripper { + return &redirectGuard{inner: inner, basePath: w.basePath} +} + +// redirectGuard is the RoundTripper the feed composes over the host's +// transport: every 3xx answered to a seam call is answered here, at the +// wire, before net/http's redirect loop can parse a Location, follow it, or +// render it into a url.Error that the operation hooks would receive whole. +// The Location is reduced to its origin on the call's refusedHop record — +// the one component the seam contract lets a redirect_refused error carry — +// then the header and the body are dropped, and the 3xx goes on as a bare +// status the generated call classifies. One mechanism covers every shape of +// the class: a foreign Location, a same-origin one, a downgraded one, a +// Location net/http could not parse, and a 3xx with none. A seam call is +// recognized by its request path — the three feed operations' own routes, +// exactly, beneath the configured base path — rather than by the per-call +// record on its context, which a host hook that returns a fresh context +// would drop; the record, when it survives, carries the origin back to the +// seam, and the strip does not depend on it. Every other request passes +// through untouched: the client's other operations — a download's +// dispatching 302 above all — keep their own redirect handling. +// +// The guard also marks a seam call's response body, so that a read that +// fails after the headers arrived — a reset stream, a connection cut +// mid-body — reaches the seam as a bodyReadError the classifier can tell +// from a body that arrived whole and did not decode. +type redirectGuard struct { + inner http.RoundTripper + basePath string +} + +func (g *redirectGuard) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := g.inner.RoundTrip(req) + if err != nil || resp == nil || !isFeedOperationPath(g.basePath, req.URL.Path) { + return resp, err + } + if !isRedirectStatus(resp.StatusCode) { + if resp.Body != nil && resp.Body != http.NoBody { + resp.Body = &markedBody{ReadCloser: resp.Body} + } + return resp, nil + } + if hop, ok := req.Context().Value(refusedHopKey{}).(*refusedHop); ok { + hop.record(locationOrigin(req.URL, resp.Header.Get("Location"))) + } + resp.Header.Del("Location") + resp.Header.Del("Content-Location") + if resp.Body != nil { + _ = resp.Body.Close() + } + resp.Body = http.NoBody + resp.ContentLength = 0 + resp.Header.Del("Content-Length") + return resp, nil +} + +// isFeedOperationPath reports a request path that is exactly one of the +// three feed operations' routes — {account}/events.json, +// {account}/inbox.json, {account}/events/stream_ticket.json — directly +// beneath the configured base path (basePath, "/"-terminated): the routes +// the seams issue, whether from a fresh cursor or a re-issued continuation. +// Anchoring at the base path and at the account segment keeps a recording's +// audit trail (/{account}/recordings/{id}/events.json) and every other +// route a host issues through the same client outside the guard. +func isFeedOperationPath(basePath, path string) bool { + rest, ok := strings.CutPrefix(path, basePath) + if !ok { + return false + } + account, route, ok := strings.Cut(rest, "/") + if !ok || account == "" || !isDigits(account) { + return false + } + return route == "events.json" || route == "inbox.json" || route == "events/stream_ticket.json" +} + +// markedBody wraps a seam call's response body so a read failure after the +// headers — as opposed to io.EOF, the body's own end — surfaces as a +// bodyReadError. +type markedBody struct{ io.ReadCloser } + +func (b *markedBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + if err != nil && !errors.Is(err, io.EOF) { + err = &bodyReadError{err: err} + } + return n, err +} + +// bodyReadError is a response body that could not be read whole: the +// connection ended, the stream was reset — a transport failure by +// construction, whatever type the HTTP stack chose for it. +type bodyReadError struct{ err error } + +func (e *bodyReadError) Error() string { + return "eventfeed: reading the response body: " + e.err.Error() +} +func (e *bodyReadError) Unwrap() error { return e.err } + +func isDigits(s string) bool { + for _, r := range s { + if r < '0' || r > '9' { + return false + } + } + return true +} + +// errHopRefused is the fixed cause a refused hop carries: never the +// Location, never the status text. +var errHopRefused = errors.New("eventfeed: refused a redirect off the API origin") + +// The two fixed tokens a redirect_refused carries in place of an origin. +// originUnparsable is §9's: the Location was absent, or yielded no complete +// origin. originUnrecorded is the other thing that can be true — the guard +// refused a hop, and the seam cannot say where to: the call's record travels +// on its context, and a host hook that returns a context unrelated to the +// one it was handed drops it. They are distinct because they say different +// things: the first is a statement about the server's header, the second +// about this connector's own bookkeeping, and an operator reading the first +// for the second would go looking for a malformed Location that never was. +const ( + originUnparsable = "unparsable" + originUnrecorded = "unrecorded" +) + +// locationOrigin reduces a refused hop's Location — resolved against the +// request's URL first, as a relative or scheme-relative reference would be +// — to its origin, the one component the seam contract lets a +// redirect_refused error carry; or the fixed token `unparsable` when the +// header is absent or yields no complete origin (§9). +func locationOrigin(base *url.URL, location string) string { + if location == "" || base == nil { + return originUnparsable + } + ref, err := url.Parse(location) + if err != nil { + return originUnparsable + } + origin, err := CanonicalOrigin(base.ResolveReference(ref).String()) + if err != nil { + return originUnparsable + } + return origin +} + +// continuationIsResume reports a followed URL that re-enters at a since — +// a 410's resume, which starts a new walk — as opposed to a position +// continuation of the walk in progress. +func continuationIsResume(pageURL string) bool { + u, err := url.Parse(pageURL) + if err != nil { + return false + } + return u.Query().Get("since") != "" +} + +// liveMinter is the TicketMinter over CreateStreamTicket. +type liveMinter struct { + svc *basecamp.EventFeedService +} + +func (m *liveMinter) MintStreamTicket(ctx context.Context) (StreamTicket, error) { + ctx, hop := withRefusedHop(ctx) + ticket, err := m.svc.CreateStreamTicket(ctx) + if err != nil { + return StreamTicket{}, mapMintError(ctx, err, hop) + } + if ticket == nil || ticket.Ticket == "" || ticket.URL == "" || ticket.ExpiresIn <= 0 { + // A malformed success: the mint answered 200 without the credential, + // the URL the connector must dial, or a positive lifetime. Nothing + // to dial and nothing a retry changes. + return StreamTicket{}, &MintError{Kind: MintUnrecoverable, Err: errors.New("the mint returned no ticket, no url, or no lifetime")} + } + return StreamTicket{Ticket: ticket.Ticket, ExpiresIn: ticket.ExpiresIn, URL: ticket.URL}, nil +} + +// mapMintError maps a CreateStreamTicket outcome onto exactly one MintErrorKind. +func mapMintError(ctx context.Context, err error, hop *refusedHop) error { + if isCancellation(ctx, err) { + return err + } + var apiErr *basecamp.Error + if errors.As(err, &apiErr) && isRedirectStatus(apiErr.HTTPStatus) { + // A mint that redirects is out of contract, and a fresh mint would + // redirect the same way: unrecoverable, with a fixed cause — the + // guard already reduced the hop to an origin nothing here renders. + _, _ = hop.get() + return &MintError{Kind: MintUnrecoverable, Err: errHopRefused} + } + if !errors.As(err, &apiErr) { + if isTransportFailure(err) { + // DNS, TLS, a dropped connection: transient, it rides the + // reconnect cycle. + return &MintError{Kind: MintTransient, Err: err} + } + // Anything else the generated call produced without a status — a + // 200 whose body did not decode, an empty response — is a + // deterministic outcome: a fresh mint answers the same way, so it + // is the malformed success §23 names. + return &MintError{Kind: MintUnrecoverable, Err: err} + } + switch { + case apiErr.HTTPStatus == http.StatusUnauthorized || apiErr.HTTPStatus == http.StatusForbidden: + return &MintError{Kind: MintUnauthorized, Err: err} + case apiErr.Retryable && apiErr.RetryAfter > 0: + // A retryable outcome exhausted inside the seam whose last response + // carried a parsed Retry-After, whatever its status (§6). Gated on + // the outcome being retryable at all: a header on a 404 names no + // wait worth taking. + return &MintError{Kind: MintThrottled, RetryAfter: time.Duration(apiErr.RetryAfter) * time.Second, Err: err} + case apiErr.Retryable || apiErr.Code == basecamp.CodeNetwork: + return &MintError{Kind: MintTransient, Err: err} + default: + return &MintError{Kind: MintUnrecoverable, Err: err} + } +} + +// isTransportFailure reports an error that says nothing about the response +// — the HTTP stack failed on the way to or from the server (a url.Error or a +// net.Error), the connection ended mid-body before the parser could read a +// whole response (io.ErrUnexpectedEOF, or a bare io.EOF from a body read), +// or the client's own resilience gate refused to send at all (the circuit +// breaker, bulkhead or rate limiter, which recover on their own clocks) — as +// opposed to one the generated call produced from a response it received +// whole, such as a body that did not decode. +func isTransportFailure(err error) bool { + if errors.Is(err, basecamp.ErrCircuitOpen) || errors.Is(err, basecamp.ErrBulkheadFull) || errors.Is(err, basecamp.ErrRateLimited) { + return true + } + var bodyErr *bodyReadError + if errors.As(err, &bodyErr) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) { + return true + } + var urlErr *url.Error + if errors.As(err, &urlErr) { + return true + } + var netErr net.Error + return errors.As(err, &netErr) +} + +// checkContinuationQuery refuses a continuation or resume URL whose query does +// not parse whole. url.URL.Query silently drops a malformed pair, and if that +// pair is `position` the re-issued operation becomes a bare present entry — +// a walk that commits a new head while skipping the history it was following. +// The URL is server-supplied text, so the error names nothing of it. +func checkContinuationQuery(rawURL string) error { + u, err := url.Parse(rawURL) + if err != nil { + return errors.New("eventfeed: the continuation URL does not parse") + } + values, err := url.ParseQuery(u.RawQuery) + if err != nil { + return errors.New("eventfeed: the continuation URL's query does not parse whole") + } + for _, v := range values { + if len(v) > 1 { + // The wrapper reads each key once (the first value); the API + // would read the last. A repeated key has no single meaning + // the seam can re-issue faithfully. + return errors.New("eventfeed: the continuation URL repeats a query key") + } + } + return nil +} + +// livePolls is the PollSource over PollEvents or PollInbox. It remembers the +// last key of the page it served so a continuation's first row can be held +// to the walk's strict order across pages; a fresh cursor starts a new walk. +type livePolls struct { + svc *basecamp.EventFeedService + lane Lane + + mu sync.Mutex + lastKey int64 +} + +func (p *livePolls) Poll(ctx context.Context, cursor Cursor, filters Filters) (PollPage, error) { + p.mu.Lock() + defer p.mu.Unlock() + if cursor.PageURL == "" || continuationIsResume(cursor.PageURL) { + // A fresh cursor, or a 410's resume (a since re-entry), starts a + // new walk; only a position continuation carries the order over. + p.lastKey = 0 + } + var page PollPage + var err error + if p.lane == InboxLane { + page, err = p.pollInbox(ctx, cursor, filters) + } else { + page, err = p.pollEvents(ctx, cursor, filters) + } + if err != nil { + return PollPage{}, err + } + if len(page.Events) > 0 { + if page.Events[0].Key() <= p.lastKey { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: fmt.Errorf("eventfeed: the continuation's first row %d does not follow the previous page", page.Events[0].Key())} + } + p.lastKey = page.Events[len(page.Events)-1].Key() + } + return page, nil +} + +func (p *livePolls) pollEvents(ctx context.Context, cursor Cursor, filters Filters) (PollPage, error) { + opts := &basecamp.PollEventsOptions{ + Since: cursor.Since, + Position: cursor.Position, + Types: filters.Types, + Buckets: filters.Buckets, + Creators: filters.Creators, + Performers: formatIDs(filters.Performers), + ExcludePerformers: formatIDs(filters.ExcludePerformers), + ActorTypes: filters.ActorTypes, + } + if cursor.PageURL != "" { + // A validated continuation or resume URL: the same operation, + // re-issued at the URL's cursor under the lane's OWN canonical + // filters. The URL's filter dimensions are not trusted — they must + // equal the lane's, or the URL is refused — and only its cursor is + // read; the connector validated the origin before this call. + if err := checkContinuationQuery(cursor.PageURL); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + parsed, err := basecamp.PollEventsOptionsFromURL(cursor.PageURL) + if err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errContinuationUnparsable} + } + if err := checkContinuationCursor(parsed.Position, parsed.Since); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + if !sameStrings(parsed.Types, opts.Types) || !sameInt64s(parsed.Buckets, opts.Buckets) || !sameInt64s(parsed.Creators, opts.Creators) || + !sameStrings(parsed.Performers, opts.Performers) || !sameStrings(parsed.ExcludePerformers, opts.ExcludePerformers) || !sameStrings(parsed.ActorTypes, opts.ActorTypes) { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errContinuationFilters} + } + opts.Since, opts.Position = parsed.Since, parsed.Position + } + ctx, hop := withRefusedHop(ctx) + page, err := p.svc.PollEvents(ctx, opts) + if err != nil { + return PollPage{}, mapPollError(ctx, err, hop, p.lane) + } + if page == nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errors.New("the poll returned no page")} + } + if err := checkNextCursor(page.Next); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + if page.Events == nil { + // The envelope's events member is required; a page without one is + // not an empty page but a malformed one, and saving its position + // would skip whatever it should have carried. + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the page carries no events member")} + } + events := make([]Event, 0, len(page.Events)) + for _, fe := range page.Events { + ev, err := eventFromFeed(fe) + if err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + if err := checkPageOrder(events, ev); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + events = append(events, ev) + } + return PollPage{Events: events, Position: page.Position, Next: page.Next}, nil +} + +func (p *livePolls) pollInbox(ctx context.Context, cursor Cursor, filters Filters) (PollPage, error) { + opts := &basecamp.PollInboxOptions{ + Since: cursor.Since, + Position: cursor.Position, + Reasons: filters.Reasons, + Types: filters.Types, + Buckets: filters.Buckets, + } + if cursor.PageURL != "" { + if err := checkContinuationQuery(cursor.PageURL); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + parsed, err := basecamp.PollInboxOptionsFromURL(cursor.PageURL) + if err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errContinuationUnparsable} + } + if err := checkContinuationCursor(parsed.Position, parsed.Since); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + if !sameStrings(parsed.Reasons, opts.Reasons) || !sameStrings(parsed.Types, opts.Types) || !sameInt64s(parsed.Buckets, opts.Buckets) { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errContinuationFilters} + } + opts.Since, opts.Position = parsed.Since, parsed.Position + } + ctx, hop := withRefusedHop(ctx) + page, err := p.svc.PollInbox(ctx, opts) + if err != nil { + return PollPage{}, mapPollError(ctx, err, hop, p.lane) + } + if page == nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errors.New("the poll returned no page")} + } + if err := checkNextCursor(page.Next); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + if page.Items == nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the page carries no items member")} + } + events := make([]Event, 0, len(page.Items)) + for _, item := range page.Items { + // The envelope's own required members, before the event's: a zero + // addressing id would become the lane's dedupe and reset key. + if item.AddressingID < 1 || item.Reason == "" || item.AddressedAt.IsZero() { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: fmt.Errorf("eventfeed: inbox item %d is missing a required member", item.AddressingID)} + } + ev, err := eventFromFeed(item.Event) + if err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + ev.Addressing = &Addressing{ID: item.AddressingID, Reason: item.Reason, AddressedAt: item.AddressedAt} + if err := checkPageOrder(events, ev); err != nil { + return PollPage{}, &PollError{Kind: PollUnrecoverable, Err: err} + } + events = append(events, ev) + } + return PollPage{Events: events, Position: page.Position, Next: page.Next}, nil +} + +// checkPageOrder holds a page to the contract's strict order of the lane's +// identity (PollPage.Events): a row whose key does not exceed the previous +// row's is a malformed page — reordered or duplicated logical deliveries +// the connector must not commit a position over. +func checkPageOrder(sofar []Event, next Event) error { + if n := len(sofar); n > 0 && next.Key() <= sofar[n-1].Key() { + return fmt.Errorf("eventfeed: poll page rows are not in strict key order at %d", next.Key()) + } + return nil +} + +// checkResumeCursor holds a 410's resume URL to the fence its body declares: +// the feed's re-enters at since=, the inbox's at since=0, and +// neither carries a position. A resume that re-enters elsewhere — since=now, +// a different id — would have an accepting handler skip retained history. +// The URL is server-supplied text, so the error names nothing of it; its +// origin is the connector's to validate before the URL is followed. +func checkResumeCursor(resume, since string) error { + u, err := url.Parse(resume) + if err != nil { + return errors.New("eventfeed: the 410's resume URL does not parse") + } + values, err := url.ParseQuery(u.RawQuery) + if err != nil { + return errors.New("eventfeed: the 410's resume URL's query does not parse whole") + } + if len(values["since"]) != 1 || values["since"][0] != since || len(values["position"]) != 0 { + return errors.New("eventfeed: the 410's resume URL does not re-enter at the fence the body declares") + } + return nil +} + +// positionRejectedMessage is the leading text of bc3's 400 for a malformed +// (or foreign-account) position — the one 400 that is recoverable by a +// since= re-entry. Every other 400 names an offending filter and is the +// configuration error a position reset cannot help. +const positionRejectedMessage = "Unrecognized position" + +// mapPollError maps a PollEvents/PollInbox outcome onto exactly one +// PollErrorKind (SPEC.md §23 "Seam Contracts"). The lane decides what a 410 +// must carry: the feed's names its epoch, the inbox's carries none. +func mapPollError(ctx context.Context, err error, hop *refusedHop, lane Lane) error { + if isCancellation(ctx, err) { + return err + } + var apiErr *basecamp.Error + if errors.As(err, &apiErr) && isRedirectStatus(apiErr.HTTPStatus) { + // A 3xx the guard answered: the origin it recorded rides as data on + // LocationOrigin, and the cause is fixed text — nothing of the + // response reaches a rendering. A refusal whose record did not reach + // the seam reports `unrecorded`, never `unparsable`: the guard + // refuses the hop off the request's route, so zero egress holds + // either way, but the two tokens answer different questions and only + // one of them is about the server's header. + origin := originUnrecorded + if recorded, ok := hop.get(); ok && recorded != "" { + origin = recorded + } + return &PollError{Kind: PollRedirectRefused, LocationOrigin: origin, Err: errHopRefused} + } + var mismatch *basecamp.FeedFilterMismatchError + if errors.As(err, &mismatch) { + if !isServerDigest(mismatch.PositionDigest) || !isServerDigest(mismatch.FiltersDigest) { + // The conflict verdict is the status; the digests are what the + // connector reports and compares. A 409 whose digests are + // missing or not bare 16-hex is not the documented conflict but + // a malformed response. (Holding the shape at the wrapper's + // decode, for every consumer, is tracked in #915.) + return &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the 409's digests are missing or malformed")} + } + return &PollError{Kind: PollFilterChanged, PositionDigest: mismatch.PositionDigest, FiltersDigest: mismatch.FiltersDigest, Err: err} + } + // The two lanes' 410s are two generated types on purpose (the feed's + // names its epoch and re-enters there; the inbox's has none and re-enters + // at since=0), so each maps in its own arm, on its own lane only, and + // its resume URL must re-enter at the fence the body declares. + var feedGone *basecamp.FeedPositionGoneError + if errors.As(err, &feedGone) { + if lane != AccountLane { + return &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the inbox answered with the feed's 410 shape")} + } + if err := checkResumeCursor(feedGone.Resume, strconv.FormatInt(feedGone.EpochAfterID, 10)); err != nil { + return &PollError{Kind: PollUnrecoverable, Err: err} + } + return &PollError{Kind: PollGone, EpochAfterID: feedGone.EpochAfterID, ResumeURL: feedGone.Resume, Err: err} + } + var inboxGone *basecamp.InboxPositionGoneError + if errors.As(err, &inboxGone) { + if lane != InboxLane { + return &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the feed answered with the inbox's 410 shape")} + } + if err := checkResumeCursor(inboxGone.Resume, "0"); err != nil { + return &PollError{Kind: PollUnrecoverable, Err: err} + } + return &PollError{Kind: PollGone, ResumeURL: inboxGone.Resume, Err: err} + } + var request *basecamp.FeedRequestError + if errors.As(err, &request) { + // The 400's reason keys the recover-versus-stop split. A server + // that sent none (it predates bc3 #13362) leaves the message as its + // only signal, and that is the fallback below — never a guess + // between the two on any other basis. + switch request.Reason { + case basecamp.FeedReasonInvalidPosition: + return &PollError{Kind: PollPositionInvalid, Msg: request.Err.Message, Err: err} + case basecamp.FeedReasonInvalidFilter: + return &PollError{Kind: PollFilterInvalid, Msg: request.Err.Message, Err: err} + case "": + // Falls through to the message fallback below. + default: + // A reason the contract does not name: the 400 is surfaced + // as undifferentiated, never guessed from its message. + return &PollError{Kind: PollUnrecoverable, Err: errors.New("eventfeed: the 400 carries a reason the contract does not name")} + } + } + if !errors.As(err, &apiErr) { + if isTransportFailure(err) { + return &PollError{Kind: PollTransient, Err: err} + } + // A 200 that did not decode, or an empty response: the unexpected + // shape §23 maps to poll_failed, since re-polling draws it again. + return &PollError{Kind: PollUnrecoverable, Err: err} + } + switch { + case apiErr.HTTPStatus == http.StatusBadRequest: + if len(apiErr.Message) >= len(positionRejectedMessage) && apiErr.Message[:len(positionRejectedMessage)] == positionRejectedMessage { + return &PollError{Kind: PollPositionInvalid, Msg: apiErr.Message, Err: err} + } + return &PollError{Kind: PollFilterInvalid, Msg: apiErr.Message, Err: err} + case apiErr.HTTPStatus == http.StatusUnauthorized || apiErr.HTTPStatus == http.StatusForbidden: + return &PollError{Kind: PollUnauthorized, Err: err} + case apiErr.Retryable && apiErr.RetryAfter > 0: + return &PollError{Kind: PollThrottled, RetryAfter: time.Duration(apiErr.RetryAfter) * time.Second, Err: err} + case apiErr.Retryable || apiErr.Code == basecamp.CodeNetwork: + return &PollError{Kind: PollTransient, Err: err} + default: + return &PollError{Kind: PollUnrecoverable, Err: err} + } +} + +// isRedirectStatus reports a 3xx: at the guard, a response to answer; at +// the seam, the status the generated call surfaced once the guard had. +func isRedirectStatus(status int) bool { + return status >= 300 && status <= 399 +} + +// isCancellation reports an end to the call driven by the CONNECTOR's own +// context, which the seam returns as-is: the connector cancelled it and reads +// the context, not a kind. Judged on the context's state, not on the error +// alone — the HTTP client's own timeout also surfaces as a wrapped +// DeadlineExceeded, and that one is a transport failure to classify. +func isCancellation(ctx context.Context, err error) bool { + return ctx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) +} + +// errContinuationUnparsable is the fixed cause for a continuation whose query +// the wrapper could not turn into options: the wrapper's own error names the +// offending value, which is server-chosen text a rendering must not carry. +var errContinuationUnparsable = errors.New("eventfeed: the continuation URL carries a filter value that does not parse") + +// checkNextCursor holds a page's `next` to a walk continuation's shape: it +// carries `position` and never `since`. A `next` that would re-enter at a +// since — a present entry, which commits the head and skips the rest of the +// walk — makes the page malformed, refused the way a positionless page is: +// no re-entry, no save. The URL is server text; the error names none of it. +func checkNextCursor(next string) error { + if next == "" { + return nil + } + u, err := url.Parse(next) + if err != nil { + return errors.New("eventfeed: the page's next URL does not parse") + } + values, err := url.ParseQuery(u.RawQuery) + if err != nil { + return errors.New("eventfeed: the page's next URL's query does not parse whole") + } + if len(values["position"]) != 1 || values["position"][0] == "" || len(values["since"]) != 0 { + return errors.New("eventfeed: the page's next URL must continue at a position, never a since") + } + return nil +} + +// errContinuationFilters is the fixed cause for a followed URL whose filter +// dimensions differ from the lane's own: the URL's filters are never +// trusted, and a continuation that would change the lineage is refused. +var errContinuationFilters = errors.New("eventfeed: the continuation URL's filters differ from the lane's") + +// sameStrings and sameInt64s compare two filter dimensions as sets — order +// and repetition aside, as the srv2 digest reads them. +func sameStrings(a, b []string) bool { return sameSet(a, b) } + +func sameInt64s(a, b []int64) bool { return sameSet(a, b) } + +func sameSet[T comparable](a, b []T) bool { + as, bs := map[T]struct{}{}, map[T]struct{}{} + for _, v := range a { + as[v] = struct{}{} + } + for _, v := range b { + bs[v] = struct{}{} + } + if len(as) != len(bs) { + return false + } + for v := range as { + if _, ok := bs[v]; !ok { + return false + } + } + return true +} + +// isServerDigest reports a bare srv2 digest: exactly 16 lowercase hex. +func isServerDigest(s string) bool { + if len(s) != 16 { + return false + } + for _, r := range s { + digit, hex := r >= '0' && r <= '9', r >= 'a' && r <= 'f' + if !digit && !hex { + return false + } + } + return true +} + +// checkContinuationCursor requires a followed URL to carry exactly one cursor +// — a position or a since — so a server URL that omits it, or spells it under +// a key the parser does not know, is never re-issued as a bare present entry +// that commits the head and skips the rest of the walk. +func checkContinuationCursor(position, since string) error { + if (position == "") == (since == "") { + return errors.New("eventfeed: the continuation URL must carry exactly one of position and since") + } + return nil +} + +// eventFromFeed maps the wrapper's FeedEvent onto the connector's Event. The +// details object passes through as the bytes the server sent, under the push +// decoder's rule — an object is kept whole, null is absent, anything else is +// refused — so the two lanes deliver byte-identical detail objects: explicit +// nulls and the members of newly cataloged types survive on both. +func eventFromFeed(fe basecamp.FeedEvent) (Event, error) { + // The generated model is value-typed, so a row missing a required member + // arrives as a zero value rather than a decode error; the push decoder + // refuses the same shapes, and a zero id would reach the dedupe ledger + // under a key nothing real can share. + if fe.ID < 1 || fe.BucketID < 1 || fe.CreatorID < 1 || fe.RecordingID < 1 || + (fe.PerformedByID != nil && *fe.PerformedByID < 1) || + fe.Kind == "" || fe.EventType == "" || fe.Action == "" || fe.CreatedAt.IsZero() { + return Event{}, fmt.Errorf("eventfeed: poll row %d is missing a required member", fe.ID) + } + ev := Event{ + ID: fe.ID, + Kind: fe.Kind, + EventType: fe.EventType, + Action: fe.Action, + CreatedAt: fe.CreatedAt, + BucketID: fe.BucketID, + CreatorID: fe.CreatorID, + PerformedByID: fe.PerformedByID, + RecordingID: fe.RecordingID, + } + if trimmed := bytes.TrimSpace(fe.Details); len(trimmed) > 0 && !bytes.Equal(trimmed, []byte("null")) { + // The push decoder's rule for the same bytes: an object, valid + // UTF-8, no lone surrogate escape — so both lanes deliver the same + // document or refuse the same one. + if !isJSONObject(trimmed) || !utf8.Valid(trimmed) || hasLoneSurrogateEscape(trimmed) { + return Event{}, fmt.Errorf("eventfeed: poll row %d carries details that are not a well-formed object", fe.ID) + } + ev.Details = json.RawMessage(slices.Clone(trimmed)) + } + return ev, nil +} + +// formatIDs renders ids for the wrapper's string-typed performer lists (which +// also admit the server's `self` literal; the connector never sends it). +func formatIDs(ids []int64) []string { + if len(ids) == 0 { + return nil + } + out := make([]string, len(ids)) + for i, id := range ids { + out[i] = strconv.FormatInt(id, 10) + } + return out +} diff --git a/go/pkg/basecamp/eventfeed/live_test.go b/go/pkg/basecamp/eventfeed/live_test.go new file mode 100644 index 000000000..34bf35e4d --- /dev/null +++ b/go/pkg/basecamp/eventfeed/live_test.go @@ -0,0 +1,1257 @@ +package eventfeed_test + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" +) + +// The Layer-1 adapters against a real generated client and a loopback API: +// every seam call is one governed operation, every outcome maps onto exactly +// one seam error kind, and a continuation is followed through the operation +// rather than fetched raw — including the redirect the tier-2 family assigns +// to this layer (conformance/event-feed README, row 15). + +type liveFixture struct { + server *httptest.Server + live *eventfeed.Live + requests atomic.Int32 + last atomic.Pointer[http.Request] +} + +func newLiveFixture(t *testing.T, lane eventfeed.Lane, handler http.HandlerFunc) *liveFixture { + t.Helper() + return newLiveFixtureWith(t, lane, handler) +} + +func newLiveFixtureWith(t *testing.T, lane eventfeed.Lane, handler http.HandlerFunc, extra ...basecamp.ClientOption) *liveFixture { + t.Helper() + f := &liveFixture{} + f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.requests.Add(1) + clone := r.Clone(r.Context()) + f.last.Store(clone) + handler(w, r) + })) + t.Cleanup(f.server.Close) + cfg := basecamp.DefaultConfig() + cfg.BaseURL = f.server.URL + opts := append([]basecamp.ClientOption{basecamp.WithMaxRetries(0), basecamp.WithBaseDelay(time.Millisecond)}, extra...) + live, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "test-token"}, "99999", lane, opts...) + if err != nil { + t.Fatalf("NewLive: %v", err) + } + f.live = live + return f +} + +func jsonResponse(w http.ResponseWriter, status int, body string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) +} + +const liveFeedPage = `{"events":[ + {"id":101,"kind":"message_created","action":"created","created_at":"2026-07-14T06:10:00.159Z","event_type":"message.created","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900}, + {"id":102,"kind":"kanban_card_moved","action":"moved","created_at":"2026-07-14T06:12:00Z","event_type":"card.moved","bucket_id":2,"creator_id":3,"performed_by_id":9007199254740993,"recording_id":901,"details":{"column_id":77,"previous_column_id":76}} +],"position":"posAAA","next":"https://3.basecampapi.com/99999/events.json?position=posAAA&types=message.created%2Ccard.moved&buckets=2"}` + +func TestLiveMinter(t *testing.T) { + t.Run("mints through CreateStreamTicket", func(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/99999/events/stream_ticket.json" { + t.Errorf("mint hit %s %s", r.Method, r.URL.Path) + } + if r.Header.Get("Authorization") != "Bearer test-token" { + t.Errorf("mint carried Authorization %q", r.Header.Get("Authorization")) + } + jsonResponse(w, 200, `{"ticket":"tkt-1","expires_in":120,"url":"wss://cable.example.test/99999?ticket=tkt-1"}`) + }) + ticket, err := f.live.Minter().MintStreamTicket(context.Background()) + if err != nil { + t.Fatalf("MintStreamTicket: %v", err) + } + if ticket.Ticket != "tkt-1" || ticket.ExpiresIn != 120 || ticket.URL != "wss://cable.example.test/99999?ticket=tkt-1" { + t.Fatalf("ticket = %+v", ticket) + } + }) + cases := []struct { + name string + status int + header string + body string + kind eventfeed.MintErrorKind + wait time.Duration + }{ + {"401 is unauthorized", 401, "", `{"error":"nope"}`, eventfeed.MintUnauthorized, 0}, + {"403 is unauthorized", 403, "", ``, eventfeed.MintUnauthorized, 0}, + {"429 with Retry-After is throttled", 429, "7", `{"error":"slow down"}`, eventfeed.MintThrottled, 7 * time.Second}, + {"503 is transient", 503, "", ``, eventfeed.MintTransient, 0}, + {"404 is unrecoverable", 404, "", `{"error":"gone"}`, eventfeed.MintUnrecoverable, 0}, + {"a Retry-After on a 404 is still unrecoverable", 404, "5", `{"error":"gone"}`, eventfeed.MintUnrecoverable, 0}, + {"a 200 that does not decode is unrecoverable", 200, "", `{"ticket": `, eventfeed.MintUnrecoverable, 0}, + {"a malformed success is unrecoverable", 200, "", `{"ticket":"","expires_in":120,"url":""}`, eventfeed.MintUnrecoverable, 0}, + {"a non-positive lifetime is a malformed success", 200, "", `{"ticket":"t","expires_in":0,"url":"wss://cable.example.test/1?ticket=t"}`, eventfeed.MintUnrecoverable, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + if tc.header != "" { + w.Header().Set("Retry-After", tc.header) + } + jsonResponse(w, tc.status, tc.body) + }) + _, err := f.live.Minter().MintStreamTicket(context.Background()) + var me *eventfeed.MintError + if !errors.As(err, &me) || me.Kind != tc.kind { + t.Fatalf("error = %v, want MintError kind %s", err, tc.kind) + } + if me.RetryAfter != tc.wait { + t.Fatalf("RetryAfter = %v, want %v", me.RetryAfter, tc.wait) + } + }) + } +} + +func TestLivePolls_EventsPageAndQuery(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, liveFeedPage) + }) + page, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{ + Types: []string{"message.created", "card.moved"}, + Buckets: []int64{2, 1}, + Creators: []int64{3}, + Performers: []int64{9}, + ExcludePerformers: []int64{77}, + ActorTypes: []string{eventfeed.ActorTypeAgent}, + }) + if err != nil { + t.Fatalf("Poll: %v", err) + } + req := f.last.Load() + if req.Method != http.MethodGet || req.URL.Path != "/99999/events.json" { + t.Fatalf("poll hit %s %s", req.Method, req.URL.Path) + } + want := url.Values{ + "position": {"pos-0"}, "types": {"message.created,card.moved"}, "buckets": {"2,1"}, "creators": {"3"}, + "performers": {"9"}, "exclude_performers": {"77"}, "actor_types": {"agent"}, + } + if got := req.URL.Query(); got.Encode() != want.Encode() { + t.Fatalf("query = %v, want %v", got, want) + } + if page.Position != "posAAA" || !strings.HasPrefix(page.Next, "https://3.basecampapi.com/99999/events.json?position=posAAA") { + t.Fatalf("page = %+v", page) + } + if len(page.Events) != 2 { + t.Fatalf("events = %d, want 2", len(page.Events)) + } + if page.Events[0].PerformedByID != nil || page.Events[0].Details != nil || page.Events[0].Addressing != nil { + t.Errorf("event 101 = %+v, want a direct action with no details and no addressing", page.Events[0]) + } + ev := page.Events[1] + if ev.PerformedByID == nil || *ev.PerformedByID != 9007199254740993 { + t.Errorf("event 102 PerformedByID = %v, want 9007199254740993", ev.PerformedByID) + } + var details struct { + ColumnID int64 `json:"column_id"` + PreviousColumnID int64 `json:"previous_column_id"` + } + if err := json.Unmarshal(ev.Details, &details); err != nil || details.ColumnID != 77 || details.PreviousColumnID != 76 { + t.Errorf("event 102 Details = %s (%v), want the columns", ev.Details, err) + } +} + +func TestLivePolls_FollowsAContinuationThroughTheOperation(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[],"position":"posBBB"}`) + }) + next := f.server.URL + "/99999/events.json?position=posAAA&types=message.created%2Ccard.moved&buckets=2" + // The lane's own filters go back, in the lane's spelling — the URL's + // filters are never trusted, only required to agree as sets. + page, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{PageURL: next}, eventfeed.Filters{Types: []string{"card.moved", "message.created"}, Buckets: []int64{2}}) + if err != nil { + t.Fatalf("Poll: %v", err) + } + req := f.last.Load() + want := url.Values{"position": {"posAAA"}, "types": {"card.moved,message.created"}, "buckets": {"2"}} + if got := req.URL.Query(); got.Encode() != want.Encode() { + t.Fatalf("continuation query = %v, want %v", got, want) + } + if page.Position != "posBBB" || page.Next != "" { + t.Fatalf("page = %+v, want the walk's end", page) + } + // A lane configured with a repeated value still matches the canonical set. + if _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{PageURL: next}, eventfeed.Filters{Types: []string{"card.moved", "message.created", "message.created"}, Buckets: []int64{2, 2}}); err != nil { + t.Fatalf("a repeated configured value against the canonical set: %v", err) + } + // A URL whose filters differ from the lane's is refused, unissued. + before := f.requests.Load() + _, err = f.live.Polls().Poll(context.Background(), eventfeed.Cursor{PageURL: next}, eventfeed.Filters{Types: []string{"message.created"}}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable || f.requests.Load() != before { + t.Fatalf("error = %v after %d requests, want unrecoverable with no request for differing filters", err, f.requests.Load()-before) + } +} + +// TestLivePolls_ARelativeLocationResolvesToItsOrigin: a relative or +// scheme-relative Location is resolved against the request before it is +// reduced to an origin, so the seam reports the origin the hop named. +func TestLivePolls_ARelativeLocationResolvesToItsOrigin(t *testing.T) { + for name, location := range map[string]string{"relative": "/maintenance", "scheme-relative": "//other.example.test/path"} { + t.Run(name, func(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", location) + w.WriteHeader(http.StatusFound) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused { + t.Fatalf("error = %v, want redirect_refused", err) + } + want := f.server.URL + if name == "scheme-relative" { + want = "http://other.example.test" + } + if pe.LocationOrigin != want { + t.Fatalf("LocationOrigin = %q, want %q", pe.LocationOrigin, want) + } + }) + } +} + +// TestLivePolls_AResumeStartsANewWalk: a 410's resume re-enters at a since, +// which is a new walk — its first page may carry keys below the previous +// walk's last page — while a position continuation carries the order over. +func TestLivePolls_AResumeStartsANewWalk(t *testing.T) { + const row = `"kind":"message_created","event_type":"message.created","action":"created","created_at":"2026-08-01T12:00:00Z","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900` + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("since") != "" { + jsonResponse(w, 200, `{"events":[{"id":50,`+row+`}],"position":"pos-r"}`) + return + } + jsonResponse(w, 200, `{"events":[{"id":100,`+row+`},{"id":102,`+row+`}],"position":"pos-1"}`) + }) + polls := f.live.Polls() + if _, err := polls.Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}); err != nil { + t.Fatalf("first walk: %v", err) + } + if _, err := polls.Poll(context.Background(), eventfeed.Cursor{PageURL: f.server.URL + "/99999/events.json?since=5"}, eventfeed.Filters{}); err != nil { + t.Fatalf("a resume's first page below the previous walk: %v", err) + } +} + +// TestLivePolls_APageWithoutItsCollectionIsMalformed: a 200 whose envelope +// omits its events (or items) member — or sets it null — is not an empty +// page but a malformed one, on either lane; an empty array is a page. +func TestLivePolls_APageWithoutItsCollectionIsMalformed(t *testing.T) { + for name, tc := range map[string]struct { + lane eventfeed.Lane + body string + kind eventfeed.PollErrorKind + ok bool + }{ + "feed absent": {eventfeed.AccountLane, `{"position":"p"}`, eventfeed.PollUnrecoverable, false}, + "feed null": {eventfeed.AccountLane, `{"events":null,"position":"p"}`, eventfeed.PollUnrecoverable, false}, + "feed empty": {eventfeed.AccountLane, `{"events":[],"position":"p"}`, 0, true}, + "inbox absent": {eventfeed.InboxLane, `{"position":"p"}`, eventfeed.PollUnrecoverable, false}, + "inbox empty": {eventfeed.InboxLane, `{"items":[],"position":"p"}`, 0, true}, + } { + t.Run(name, func(t *testing.T) { + f := newLiveFixture(t, tc.lane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, tc.body) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + if tc.ok { + if err != nil { + t.Fatalf("an empty page: %v", err) + } + return + } + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != tc.kind { + t.Fatalf("error = %v, want %s", err, tc.kind) + } + }) + } +} + +// TestLivePolls_ANextMustContinueAtAPosition: a page whose `next` would +// re-enter at a since — a present entry — is malformed, refused the way a +// positionless page is, on either lane. +func TestLivePolls_ANextMustContinueAtAPosition(t *testing.T) { + for name, tc := range map[string]struct { + lane eventfeed.Lane + body string + }{ + "feed next with since": {eventfeed.AccountLane, `{"events":[],"position":"p","next":"https://3.basecampapi.com/99999/events.json?since=now"}`}, + "feed next without position": {eventfeed.AccountLane, `{"events":[],"position":"p","next":"https://3.basecampapi.com/99999/events.json?types=message.created"}`}, + "inbox next with since": {eventfeed.InboxLane, `{"items":[],"position":"p","next":"https://3.basecampapi.com/99999/inbox.json?since=0"}`}, + } { + t.Run(name, func(t *testing.T) { + f := newLiveFixture(t, tc.lane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, tc.body) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable for a next that is not a walk continuation", err) + } + }) + } +} + +func TestLivePolls_ErrorMatrix(t *testing.T) { + cases := []struct { + name string + lane eventfeed.Lane + status int + header string + body string + check func(t *testing.T, pe *eventfeed.PollError) + }{ + {"400 position is position_invalid", eventfeed.AccountLane, 400, "", `{"error":"Unrecognized position. Resume with since= or since=now."}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollPositionInvalid { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"400 filter is filter_invalid with the message", eventfeed.AccountLane, 400, "", `{"error":"Unknown type nope.created. Fix the filters; a position reset won't help."}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollFilterInvalid || !strings.HasPrefix(pe.Msg, "Unknown type nope.created") { + t.Fatalf("pe = %+v", pe) + } + }}, + {"400 with reason invalid_position is position_invalid whatever the message", eventfeed.AccountLane, 400, "", `{"error":"anything at all","reason":"invalid_position"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollPositionInvalid { + t.Fatalf("kind = %s", pe.Kind) + } + }}, {"400 with reason invalid_filter is filter_invalid whatever the message", eventfeed.AccountLane, 400, "", `{"error":"Unrecognized position, said a server that means the filter","reason":"invalid_filter"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollFilterInvalid { + t.Fatalf("kind = %s", pe.Kind) + } + }}, {"inbox 400 with reason invalid_position is position_invalid", eventfeed.InboxLane, 400, "", `{"error":"bad","reason":"invalid_position"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollPositionInvalid { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"409 with a malformed digest is unrecoverable", eventfeed.AccountLane, 409, "", `{"error":"conflict","position_digest":"38b223c13c89dc89","filters_digest":"x"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"400 with an unknown reason is surfaced, never guessed", eventfeed.AccountLane, 400, "", `{"error":"Unrecognized position. Resume with since=.","reason":"invalid_something"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"409 missing a digest is unrecoverable", eventfeed.AccountLane, 409, "", `{"error":"conflict","position_digest":"38b223c13c89dc89"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, {"410 whose resume does not re-enter at the epoch is unrecoverable", eventfeed.AccountLane, 410, "", `{"error":"gone","epoch_after_id":1071915000,"resume":"https://3.basecampapi.com/99999/events.json?since=now"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, {"inbox 410 whose resume does not re-enter at 0 is unrecoverable", eventfeed.InboxLane, 410, "", `{"error":"gone","resume":"https://3.basecampapi.com/99999/inbox.json?since=5"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, {"410 whose resume carries a position is unrecoverable", eventfeed.AccountLane, 410, "", `{"error":"gone","epoch_after_id":7,"resume":"https://3.basecampapi.com/99999/events.json?since=7&position=p"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"409 is filter_changed with both digests", eventfeed.AccountLane, 409, "", `{"error":"Positions are bound to the filter set they were minted for.","position_digest":"38b223c13c89dc89","filters_digest":"44136fa355b3678a"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollFilterChanged || pe.PositionDigest != "38b223c13c89dc89" || pe.FiltersDigest != "44136fa355b3678a" { + t.Fatalf("pe = %+v", pe) + } + }}, + {"410 is gone with the epoch and resume", eventfeed.AccountLane, 410, "", `{"error":"That position predates this feed's epoch.","epoch_after_id":1071915000,"resume":"https://3.basecampapi.com/99999/events.json?since=1071915000"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollGone || pe.EpochAfterID != 1071915000 || pe.ResumeURL != "https://3.basecampapi.com/99999/events.json?since=1071915000" { + t.Fatalf("pe = %+v", pe) + } + }}, + {"inbox 410 is gone with no epoch", eventfeed.InboxLane, 410, "", `{"error":"That position predates the inbox's retention window.","resume":"https://3.basecampapi.com/99999/inbox.json?since=0"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollGone || pe.EpochAfterID != 0 || pe.ResumeURL != "https://3.basecampapi.com/99999/inbox.json?since=0" { + t.Fatalf("pe = %+v", pe) + } + }}, + {"401 is unauthorized", eventfeed.AccountLane, 401, "", `{"error":"nope"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnauthorized { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"inbox 403 for a person is unauthorized", eventfeed.InboxLane, 403, "", ``, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnauthorized { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"429 with Retry-After is throttled", eventfeed.AccountLane, 429, "3", `{"error":"slow down"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollThrottled || pe.RetryAfter != 3*time.Second { + t.Fatalf("pe = %+v", pe) + } + }}, + {"503 is transient", eventfeed.AccountLane, 503, "", ``, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollTransient { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"404 is unrecoverable", eventfeed.AccountLane, 404, "", `{"error":"no such feed"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s", pe.Kind) + } + }}, + {"a Retry-After on a 404 does not make it throttled", eventfeed.AccountLane, 404, "5", `{"error":"no such feed"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable || pe.RetryAfter != 0 { + t.Fatalf("pe = %+v, want unrecoverable with no wait", pe) + } + }}, + {"a 200 that does not decode is unrecoverable", eventfeed.AccountLane, 200, "", `{"events": [`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s, want unrecoverable (re-polling draws the same body)", pe.Kind) + } + }}, + {"a row missing required members is unrecoverable", eventfeed.AccountLane, 200, "", `{"events":[{"id":0,"kind":"","event_type":"message.created","action":"created","created_at":"2026-07-14T06:10:00Z","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900}],"position":"p"}`, + func(t *testing.T, pe *eventfeed.PollError) { + if pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("kind = %s, want unrecoverable", pe.Kind) + } + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newLiveFixture(t, tc.lane, func(w http.ResponseWriter, r *http.Request) { + if tc.header != "" { + w.Header().Set("Retry-After", tc.header) + } + jsonResponse(w, tc.status, tc.body) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) { + t.Fatalf("error = %T (%v), want *PollError", err, err) + } + tc.check(t, pe) + if got := f.requests.Load(); got != 1 { + t.Fatalf("requests = %d, want exactly one (no retry outside the seam's own budget)", got) + } + }) + } +} + +// TestLivePolls_RefusesAContinuationWhoseQueryDoesNotParse: a `next` whose +// query holds a malformed pair would have that pair silently dropped by +// url.URL.Query — and if it is `position`, the re-issued poll becomes a bare +// present entry that skips the history it was following. +func TestLivePolls_RefusesAContinuationWhoseQueryDoesNotParse(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[],"position":"posBBB"}`) + }) + next := f.server.URL + "/99999/events.json?position=pos%ZZ&types=message.created" + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{PageURL: next}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable", err) + } + if f.requests.Load() != 0 { + t.Fatalf("requests = %d, want none for a continuation that does not parse whole", f.requests.Load()) + } + if strings.Contains(pe.Error(), "pos%ZZ") { + t.Fatalf("PollError renders the continuation: %s", pe.Error()) + } +} + +// TestLivePolls_RefusesAContinuationWithoutACursor: a followed URL that +// carries filters but no position or since would re-issue as a bare present +// entry; a value the wrapper cannot parse yields a fixed cause, never the value. +func TestLivePolls_RefusesAContinuationWithoutACursor(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[],"position":"p"}`) + }) + for name, next := range map[string]string{ + "no cursor": f.server.URL + "/99999/events.json?types=message.created", + "repeated key": f.server.URL + "/99999/events.json?position=pos-new&position=pos-old", + "both cursors": f.server.URL + "/99999/events.json?position=p&since=now", + "bad filter id": f.server.URL + "/99999/events.json?position=p&buckets=leaked-secret", + } { + t.Run(name, func(t *testing.T) { + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{PageURL: next}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable", err) + } + if strings.Contains(pe.Error(), "leaked-secret") { + t.Fatalf("PollError renders the continuation's value: %s", pe.Error()) + } + }) + } + if f.requests.Load() != 0 { + t.Fatalf("requests = %d, want none", f.requests.Load()) + } +} + +// TestLiveClient_DownloadsKeepTheirDispatchingRedirect: the guard answers +// only the seams' own calls, so the client NewLive built stays a full client +// for the host — a download's authenticated first hop still 302s to the +// signed URL and the SDK follows its own dispatch. +func TestLiveClient_DownloadsKeepTheirDispatchingRedirect(t *testing.T) { + var f *liveFixture + f = newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/99999/blobs/1/download": + http.Redirect(w, r, f.server.URL+"/signed/blob", http.StatusFound) + case "/signed/blob": + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write([]byte("PNGDATA")) + default: + http.NotFound(w, r) + } + }) + result, err := f.live.Client().ForAccount("99999").DownloadURL(context.Background(), f.server.URL+"/99999/blobs/1/download") + if err != nil { + t.Fatalf("DownloadURL through the feed's client: %v", err) + } + defer result.Body.Close() + data, _ := io.ReadAll(result.Body) + if string(data) != "PNGDATA" { + t.Fatalf("downloaded %q, want the signed hop's bytes", data) + } +} + +// TestLivePolls_A410OfTheOtherLanesShapeIsMalformed: the feed's 410 names +// its epoch and the inbox's carries none; a body of the other lane's shape +// is unrecoverable rather than a gap with a fence that does not exist. +// +// What refuses the other lane's body is its resume — the fence, which is the +// lane's own re-entry — and the cases say which of the two members they are +// about: `epochWithFeedFence` carries both the epoch and the feed's fence, +// `epochWithInboxFence` isolates the stray member from the fence. +func TestLivePolls_A410OfTheOtherLanesShapeIsMalformed(t *testing.T) { + const epochWithFeedFence = `{"error":"gone","epoch_after_id":500,"resume":"https://3.basecampapi.com/99999/events.json?since=500"}` + const inboxFence = `{"error":"gone","resume":"https://3.basecampapi.com/99999/inbox.json?since=0"}` + const epochWithInboxFence = `{"error":"gone","epoch_after_id":500,"resume":"https://3.basecampapi.com/99999/inbox.json?since=0"}` + for _, tc := range []struct { + name string + lane eventfeed.Lane + body string + kind eventfeed.PollErrorKind + epoch int64 + resume string + }{ + {"feed with its epoch and fence", eventfeed.AccountLane, epochWithFeedFence, eventfeed.PollGone, 500, "https://3.basecampapi.com/99999/events.json?since=500"}, + {"feed without an epoch", eventfeed.AccountLane, inboxFence, eventfeed.PollUnrecoverable, 0, ""}, + {"inbox with its own fence", eventfeed.InboxLane, inboxFence, eventfeed.PollGone, 0, "https://3.basecampapi.com/99999/inbox.json?since=0"}, + {"inbox with the feed's fence", eventfeed.InboxLane, epochWithFeedFence, eventfeed.PollUnrecoverable, 0, ""}, + // The decision this case pins: on the inbox lane the fence is the + // control, and an epoch_after_id the inbox contract does not declare + // is inert — the generated 410 shape has no member to decode it into + // and the inbox arm sets no epoch, so nothing it could say is + // reachable. Refusing the body on the stray member's presence would + // turn a gap that recovers correctly at since=0 into an + // unrecoverable one, which is worse for the case it claims to guard. + // Holding a response to the members its shape declares belongs at + // the wrapper's decode, for every consumer and every response, and + // is tracked separately in #915. + {"inbox with a stray epoch and its own fence", eventfeed.InboxLane, epochWithInboxFence, eventfeed.PollGone, 0, "https://3.basecampapi.com/99999/inbox.json?since=0"}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newLiveFixture(t, tc.lane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 410, tc.body) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != tc.kind { + t.Fatalf("error = %v, want %s", err, tc.kind) + } + if tc.kind == eventfeed.PollGone && (pe.EpochAfterID != tc.epoch || pe.ResumeURL != tc.resume) { + t.Fatalf("gone carries epoch %d resume %q, want %d and %q", pe.EpochAfterID, pe.ResumeURL, tc.epoch, tc.resume) + } + }) + } +} + +// TestLivePolls_TheGuardRecognizesRoutesBeneathABasePath: a base URL with a +// path prefix puts the feed routes beneath it; the guard still answers the +// 3xx, and the foreign origin sees nothing. +func TestLivePolls_TheGuardRecognizesRoutesBeneathABasePath(t *testing.T) { + var sentinelHits atomic.Int32 + sentinel := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sentinelHits.Add(1) + jsonResponse(w, 200, `{"events":[],"position":"stolen"}`) + })) + t.Cleanup(sentinel.Close) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/99999/events.json" { + t.Errorf("request path %q, want the route beneath the base path", r.URL.Path) + } + w.Header().Set("Location", sentinel.URL+"/99999/events.json?position=pos-0&token=leak") + w.WriteHeader(http.StatusFound) + })) + t.Cleanup(server.Close) + cfg := basecamp.DefaultConfig() + cfg.BaseURL = server.URL + "/api/v1" + live, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "t"}, "99999", eventfeed.AccountLane, basecamp.WithMaxRetries(0)) + if err != nil { + t.Fatalf("NewLive: %v", err) + } + _, err = live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused || pe.LocationOrigin != sentinel.URL { + t.Fatalf("error = %v, want redirect_refused carrying %q", err, sentinel.URL) + } + if sentinelHits.Load() != 0 { + t.Fatalf("the foreign origin received %d request(s), want zero egress", sentinelHits.Load()) + } +} + +// TestNewLive_AnEscapedBasePathIsRefusedRatherThanMisanchored: the guard is +// anchored at the base URL's DECODED path and the client sends the escaped +// one, so a base URL whose two forms differ anchors the guard somewhere the +// requests never go — `/api%2f` reads as `/api/` here and goes on the wire +// as `/api%2f/...`, where the guard finds no account segment, passes the 3xx +// through, and the HTTP stack follows it off the API origin with the +// request's query. NewLive refuses that spelling and every other one, and +// the base paths it still accepts are ones the guard covers: each answers +// its 302 at the wire with zero egress. +func TestNewLive_AnEscapedBasePathIsRefusedRatherThanMisanchored(t *testing.T) { + for _, suffix := range []string{"/api%2f", "/api%2f/", "/a%2Eb", "/%2e%2e/api"} { + t.Run("refused"+suffix, func(t *testing.T) { + cfg := basecamp.DefaultConfig() + cfg.BaseURL = "https://3.basecampapi.com" + suffix + if _, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "t"}, "99999", eventfeed.AccountLane); err == nil { + t.Fatalf("NewLive accepted the base path %q, which anchors the guard where the requests do not go", suffix) + } + }) + } + // The boundary the refusal must not overrun: a percent-encoding whose + // escaped form IS the canonical encoding of its decoded path anchors the + // guard correctly, and still refuses the hop with zero egress. + for _, suffix := range []string{"", "/api/v1", "/a%20b"} { + t.Run("covered"+suffix, func(t *testing.T) { + var sentinelHits atomic.Int32 + sentinel := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sentinelHits.Add(1) + jsonResponse(w, 200, `{"events":[],"position":"stolen"}`) + })) + t.Cleanup(sentinel.Close) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", sentinel.URL+"/99999/events.json?position=pos-0&token=leak") + w.WriteHeader(http.StatusFound) + })) + t.Cleanup(server.Close) + cfg := basecamp.DefaultConfig() + cfg.BaseURL = server.URL + suffix + live, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "t"}, "99999", eventfeed.AccountLane, basecamp.WithMaxRetries(0)) + if err != nil { + t.Fatalf("NewLive refused the base path %q: %v", suffix, err) + } + _, err = live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused || pe.LocationOrigin != sentinel.URL { + t.Fatalf("error = %v, want redirect_refused carrying %q", err, sentinel.URL) + } + if sentinelHits.Load() != 0 { + t.Fatalf("the foreign origin received %d request(s), want zero egress", sentinelHits.Load()) + } + }) + } +} + +// TestRedirectGuard_MatchesOnlyTheFeedRoutes: the guard claims exactly the +// three feed routes beneath the configured base path — never a recording's +// audit trail, whose path also ends in {id}/events.json, and never a route +// outside the base path. +func TestRedirectGuard_MatchesOnlyTheFeedRoutes(t *testing.T) { + for _, tc := range []struct { + base, path string + want bool + }{ + {"/", "/99999/events.json", true}, + {"/", "/99999/inbox.json", true}, + {"/", "/99999/events/stream_ticket.json", true}, + {"/api/v1/", "/api/v1/99999/events.json", true}, + {"/", "/api/v1/99999/events.json", false}, + {"/api/v1/", "/99999/events.json", false}, + {"/", "/99999/recordings/12345/events.json", false}, + {"/", "/99999/buckets/2085958499/recordings/12345/events.json", false}, + {"/", "/abc/events.json", false}, + {"/", "/99999/events.json/extra", false}, + } { + if got := eventfeed.ExportIsFeedOperationPath(tc.base, tc.path); got != tc.want { + t.Errorf("isFeedOperationPath(%q, %q) = %v, want %v", tc.base, tc.path, got, tc.want) + } + } +} + +// TestLive_AResetBodyIsTransient: a body whose read fails after the headers +// — whatever error type the HTTP stack chose — is a transport failure on +// both seams, told apart from a whole body that did not decode. +func TestLive_AResetBodyIsTransient(t *testing.T) { + f := newLiveFixtureWith(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[],"position":"p"}`) + }, basecamp.WithTransport(resettingBodyTransport{inner: http.DefaultTransport})) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollTransient { + t.Fatalf("poll error = %v, want transient for a reset body", err) + } + _, err = f.live.Minter().MintStreamTicket(context.Background()) + var me *eventfeed.MintError + if !errors.As(err, &me) || me.Kind != eventfeed.MintTransient { + t.Fatalf("mint error = %v, want transient for a reset body", err) + } +} + +// resettingBodyTransport answers every request with a body whose read fails +// with a plain error after the headers — the shape of an HTTP/2 stream +// reset, which is neither a url.Error nor a net.Error. +type resettingBodyTransport struct{ inner http.RoundTripper } + +func (t resettingBodyTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.inner.RoundTrip(req) + if err != nil { + return nil, err + } + _ = resp.Body.Close() + resp.Body = io.NopCloser(resetReader{}) + return resp, nil +} + +type resetReader struct{} + +func (resetReader) Read([]byte) (int, error) { return 0, errors.New("stream error: RST_STREAM") } + +// TestLivePolls_AContinuationOutOfOrderIsMalformed: strict order holds +// across a walk's pages — a continuation whose first row does not follow the +// previous page's last is unrecoverable, while a fresh cursor starts over. +func TestLivePolls_AContinuationOutOfOrderIsMalformed(t *testing.T) { + const row = `"kind":"message_created","event_type":"message.created","action":"created","created_at":"2026-08-01T12:00:00Z","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900` + var f *liveFixture + f = newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Query().Get("position") { + case "pos-0": + jsonResponse(w, 200, `{"events":[{"id":100,`+row+`},{"id":102,`+row+`}],"position":"pos-1","next":"`+f.server.URL+`/99999/events.json?position=pos-cont"}`) + default: + jsonResponse(w, 200, `{"events":[{"id":101,`+row+`}],"position":"pos-2"}`) + } + }) + polls := f.live.Polls() + page, err := polls.Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + if err != nil || page.Next == "" { + t.Fatalf("first page = %+v, %v", page, err) + } + // A second connector's source over the same binding walks on its own. + if _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-fresh"}, eventfeed.Filters{}); err != nil { + t.Fatalf("another source's fresh walk: %v", err) + } + _, err = polls.Poll(context.Background(), eventfeed.Cursor{PageURL: page.Next}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("continuation error = %v, want unrecoverable for a row behind the previous page", err) + } + if _, err := polls.Poll(context.Background(), eventfeed.Cursor{Position: "pos-fresh"}, eventfeed.Filters{}); err != nil { + t.Fatalf("a fresh cursor after the refused continuation: %v", err) + } +} + +// TestLivePolls_APageOutOfOrderIsMalformed: rows arrive in strict order of +// the lane's identity; a page that repeats or reorders keys is unrecoverable +// on either lane. +func TestLivePolls_APageOutOfOrderIsMalformed(t *testing.T) { + const row = `"kind":"message_created","event_type":"message.created","action":"created","created_at":"2026-08-01T12:00:00Z","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900` + for name, tc := range map[string]struct { + lane eventfeed.Lane + body string + }{ + "feed reordered": {eventfeed.AccountLane, `{"events":[{"id":102,` + row + `},{"id":101,` + row + `}],"position":"p"}`}, + "feed duplicated": {eventfeed.AccountLane, `{"events":[{"id":101,` + row + `},{"id":101,` + row + `}],"position":"p"}`}, + "inbox reordered": {eventfeed.InboxLane, `{"items":[{"addressing_id":12,"reason":"mentioned","addressed_at":"2026-08-01T12:00:01Z","event":{"id":101,` + row + `}},{"addressing_id":11,"reason":"assigned","addressed_at":"2026-08-01T12:00:01Z","event":{"id":102,` + row + `}}],"position":"p"}`}, + } { + t.Run(name, func(t *testing.T) { + f := newLiveFixture(t, tc.lane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, tc.body) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable for a page out of order", err) + } + }) + } +} + +// TestLivePolls_TheGuardSurvivesAHookThatReplacesTheContext: a host hook that +// returns a fresh context from OnRequestStart drops the seam's per-call +// record; the guard still answers the 3xx, because it recognizes the seam +// call by its route, and the foreign origin still sees nothing. What the +// dropped record costs is the origin, and the seam says so: `unrecorded`, +// never `unparsable` — an operator reading the §9 token would go looking for +// a Location the server never malformed. +func TestLivePolls_TheGuardSurvivesAHookThatReplacesTheContext(t *testing.T) { + var sentinelHits atomic.Int32 + sentinel := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sentinelHits.Add(1) + jsonResponse(w, 200, `{"events":[],"position":"stolen"}`) + })) + t.Cleanup(sentinel.Close) + f := newLiveFixtureWith(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", sentinel.URL+"/99999/events.json?position=pos-0&token=leak") + w.WriteHeader(http.StatusFound) + }, basecamp.WithHooks(contextDroppingHooks{})) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused { + t.Fatalf("error = %v, want redirect_refused", err) + } + if strings.Contains(pe.Error(), "leak") { + t.Fatalf("PollError renders the refused Location: %s", pe.Error()) + } + if pe.LocationOrigin != "unrecorded" { + t.Fatalf("LocationOrigin = %q, want %q for a refusal whose record the hook dropped", pe.LocationOrigin, "unrecorded") + } + if sentinelHits.Load() != 0 { + t.Fatalf("the foreign origin received %d request(s), want zero egress", sentinelHits.Load()) + } +} + +// contextDroppingHooks returns a fresh context from every start hook — the +// misuse a guard keyed on context values would not survive. +type contextDroppingHooks struct{ basecamp.NoopHooks } + +func (contextDroppingHooks) OnOperationStart(context.Context, basecamp.OperationInfo) context.Context { + return context.Background() +} + +func (contextDroppingHooks) OnRequestStart(context.Context, basecamp.RequestInfo) context.Context { + return context.Background() +} + +// TestLive_ATruncatedBodyIsTransient: a connection that ends after the +// headers, mid-body, is a transport failure to retry — on the mint and on the +// poll — not a malformed response to terminate on. +func TestLive_ATruncatedBodyIsTransient(t *testing.T) { + truncate := func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Length", "4096") + w.WriteHeader(200) + _, _ = w.Write([]byte(`{"events":[`)) + } + f := newLiveFixture(t, eventfeed.AccountLane, truncate) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollTransient { + t.Fatalf("poll error = %v, want transient for a truncated body", err) + } + _, err = f.live.Minter().MintStreamTicket(context.Background()) + var me *eventfeed.MintError + if !errors.As(err, &me) || me.Kind != eventfeed.MintTransient { + t.Fatalf("mint error = %v, want transient for a truncated body", err) + } +} + +// TestLivePolls_DetailsMatchThePushLaneByteForByte: a row's details object +// reaches the connector as the bytes the server sent on both lanes — an +// explicit null member and a member of a type this SDK does not model +// survive the poll adapter exactly as they survive the push decoder — and a +// details value that is not an object is refused as the push lane refuses it. +func TestLivePolls_DetailsMatchThePushLaneByteForByte(t *testing.T) { + const row = `{"id":7001,"kind":"boost_created","event_type":"boost.created","action":"created","created_at":"2026-08-01T12:00:00Z","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900,"actor_type":"user","visible_to_clients":true,"details":{"boost_id":5,"boosted_event_id":null,"boosted_event_type":null,"future_member":{"nested":[1,"two",null]}}}` + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[`+row+`],"position":"pos-1"}`) + }) + page, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + if err != nil || len(page.Events) != 1 { + t.Fatalf("Poll = %+v, %v", page, err) + } + pushed, err := eventfeed.ExportDecodePushEvent([]byte(row)) + if err != nil { + t.Fatalf("push decode: %v", err) + } + if !bytes.Equal(page.Events[0].Details, pushed.Details) { + t.Fatalf("poll details %s\nwant the push lane's %s", page.Events[0].Details, pushed.Details) + } + if !strings.Contains(string(page.Events[0].Details), `"boosted_event_id":null`) || !strings.Contains(string(page.Events[0].Details), `"future_member"`) { + t.Fatalf("details lost a null or an unmodeled member: %s", page.Events[0].Details) + } + for name, details := range map[string]string{"a null is absent": "null", "absent is absent": ""} { + t.Run(name, func(t *testing.T) { + body := strings.Replace(row, `,"details":{"boost_id":5,"boosted_event_id":null,"boosted_event_type":null,"future_member":{"nested":[1,"two",null]}}`, "", 1) + if details != "" { + body = strings.Replace(body, `"recording_id":900`, `"recording_id":900,"details":`+details, 1) + } + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[`+body+`],"position":"pos-1"}`) + }) + page, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + if err != nil || len(page.Events) != 1 || page.Events[0].Details != nil { + t.Fatalf("Poll = %+v, %v; want one event with no details", page, err) + } + }) + } + for name, details := range map[string]string{ + "invalid utf-8": `{"note":"\xff"}`, + "lone surrogate escape": `{"note":"\ud800"}`, + } { + t.Run(name+" is refused as the push lane refuses it", func(t *testing.T) { + body := strings.Replace(row, `{"boost_id":5,"boosted_event_id":null,"boosted_event_type":null,"future_member":{"nested":[1,"two",null]}}`, details, 1) + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[`+body+`],"position":"pos-1"}`) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable", err) + } + }) + } + t.Run("a non-object is refused", func(t *testing.T) { + body := strings.Replace(row, `{"boost_id":5,"boosted_event_id":null,"boosted_event_type":null,"future_member":{"nested":[1,"two",null]}}`, `7`, 1) + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[`+body+`],"position":"pos-1"}`) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable for non-object details", err) + } + }) +} + +// TestLivePolls_AMalformedLocationIsARefusedHop: a 3xx whose Location does +// not parse is answered by the guard before net/http's redirect loop would +// parse it into a url.Error; the seam classifies it as a refused hop, not a +// transport failure to retry into, and no rendering — the seam's error or the +// operation hooks' — carries the header. +func TestLivePolls_AMalformedLocationIsARefusedHop(t *testing.T) { + hooks := &recordingHooks{} + f := newLiveFixtureWith(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", "http://[::1]:namedport/leak") + w.WriteHeader(http.StatusFound) + }, basecamp.WithHooks(hooks)) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused || pe.LocationOrigin != "unparsable" { + t.Fatalf("error = %v, want redirect_refused with the unparsable token", err) + } + if strings.Contains(pe.Error(), "leak") { + t.Fatalf("PollError renders the malformed Location: %s", pe.Error()) + } + hooks.assertNoLeak(t, "leak") +} + +// recordingHooks captures what the operation hooks are handed, so a test can +// assert that the refused Location reached none of them. +type recordingHooks struct { + basecamp.NoopHooks + mu sync.Mutex + renderings []string +} + +func (h *recordingHooks) OnOperationEnd(ctx context.Context, op basecamp.OperationInfo, err error, d time.Duration) { + if err != nil { + h.mu.Lock() + h.renderings = append(h.renderings, err.Error()) + h.mu.Unlock() + } +} + +func (h *recordingHooks) OnRequestEnd(ctx context.Context, info basecamp.RequestInfo, result basecamp.RequestResult) { + h.mu.Lock() + defer h.mu.Unlock() + if result.Error != nil { + h.renderings = append(h.renderings, result.Error.Error()) + } + h.renderings = append(h.renderings, fmt.Sprintf("%+v", result)) +} + +func (h *recordingHooks) assertNoLeak(t *testing.T, token string) { + t.Helper() + h.mu.Lock() + defer h.mu.Unlock() + if len(h.renderings) == 0 { + t.Fatalf("the hooks recorded nothing; the test observes no rendering") + } + for _, r := range h.renderings { + if strings.Contains(r, token) { + t.Fatalf("a hook rendering carries the refused Location: %s", r) + } + } +} + +// TestLivePolls_TheClientsOwnTimeoutIsTransient: the HTTP client's timeout +// surfaces as a wrapped DeadlineExceeded while the connector's context is +// live; that is a transport failure to retry, not the connector's cancellation. +func TestLivePolls_TheClientsOwnTimeoutIsTransient(t *testing.T) { + f := &liveFixture{} + f.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + })) + t.Cleanup(f.server.Close) + cfg := basecamp.DefaultConfig() + cfg.BaseURL = f.server.URL + live, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "t"}, "99999", eventfeed.AccountLane, + basecamp.WithMaxRetries(0), basecamp.WithTimeout(50*time.Millisecond)) + if err != nil { + t.Fatalf("NewLive: %v", err) + } + _, err = live.Polls().Poll(context.Background(), eventfeed.Cursor{Since: "now"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollTransient { + t.Fatalf("error = %v, want transient for the client's own timeout", err) + } +} + +func TestLivePolls_GateRefusalsAreTransient(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"events":[],"position":"p"}`) + }) + // The seam's mapping is exercised directly: a gate refusal never reaches + // the wire, so there is no response to script. + for _, sentinel := range []error{basecamp.ErrCircuitOpen, basecamp.ErrBulkheadFull, basecamp.ErrRateLimited} { + if kind := eventfeed.ExportMapPollErrorKind(sentinel); kind != eventfeed.PollTransient { + t.Errorf("%v mapped to %s, want transient", sentinel, kind) + } + if kind := eventfeed.ExportMapMintErrorKind(sentinel); kind != eventfeed.MintTransient { + t.Errorf("%v mapped to %s, want transient", sentinel, kind) + } + } + _ = f +} + +func TestLivePolls_InboxItems(t *testing.T) { + f := newLiveFixture(t, eventfeed.InboxLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"items":[{"addressing_id":991,"reason":"mentioned","addressed_at":"2026-07-14T06:10:00Z","event":{"id":101,"kind":"comment_created","action":"created","created_at":"2026-07-14T06:10:00Z","event_type":"comment.created","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900}}],"position":"ipos-1"}`) + }) + page, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Since: "0"}, eventfeed.Filters{ + Reasons: []string{"mentioned", "assigned"}, Types: []string{"comment.created"}, Buckets: []int64{2}, + }) + if err != nil { + t.Fatalf("Poll: %v", err) + } + req := f.last.Load() + if req.URL.Path != "/99999/inbox.json" { + t.Fatalf("inbox poll hit %s", req.URL.Path) + } + want := url.Values{"since": {"0"}, "reasons": {"mentioned,assigned"}, "types": {"comment.created"}, "buckets": {"2"}} + if got := req.URL.Query(); got.Encode() != want.Encode() { + t.Fatalf("query = %v, want %v", got, want) + } + if len(page.Events) != 1 || page.Events[0].Addressing == nil || page.Events[0].Addressing.ID != 991 || + page.Events[0].Addressing.Reason != "mentioned" || page.Events[0].Key() != 991 || page.Events[0].ID != 101 { + t.Fatalf("items = %+v, want one item keyed by addressing id 991 over event 101", page.Events) + } +} + +func TestLivePolls_RefusesAnInboxItemMissingItsEnvelope(t *testing.T) { + f := newLiveFixture(t, eventfeed.InboxLane, func(w http.ResponseWriter, r *http.Request) { + jsonResponse(w, 200, `{"items":[{"reason":"mentioned","addressed_at":"2026-07-14T06:10:00Z","event":{"id":101,"kind":"comment_created","action":"created","created_at":"2026-07-14T06:10:00Z","event_type":"comment.created","bucket_id":2,"creator_id":3,"performed_by_id":null,"recording_id":900}}],"position":"ipos-1"}`) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Since: "0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollUnrecoverable { + t.Fatalf("error = %v, want unrecoverable for an item with no addressing_id", err) + } +} + +// TestLivePolls_RefusesACrossOriginRedirectWithZeroEgress is the Layer-1 302 +// test the tier-2 family assigns to this layer: a validated same-origin +// continuation answers 302 with a foreign Location. The guard answers the +// hop at the wire — the sentinel server behind the Location never sees a +// request — and the seam reports redirect_refused carrying the Location's +// origin, while the operation hooks see neither the Location nor its query. +func TestLivePolls_RefusesACrossOriginRedirectWithZeroEgress(t *testing.T) { + var sentinelHits atomic.Int32 + sentinel := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sentinelHits.Add(1) + jsonResponse(w, 200, `{"events":[],"position":"stolen"}`) + })) + t.Cleanup(sentinel.Close) + hooks := &recordingHooks{} + f := newLiveFixtureWith(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", sentinel.URL+"/99999/events.json?position=pos-0&token=leak") + jsonResponse(w, http.StatusFound, `{"error":"moved to leak"}`) + }, basecamp.WithHooks(hooks)) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused { + t.Fatalf("error = %v, want redirect_refused", err) + } + if pe.LocationOrigin != sentinel.URL { + t.Fatalf("LocationOrigin = %q, want the refused origin %q", pe.LocationOrigin, sentinel.URL) + } + if strings.Contains(pe.Error(), "leak") || strings.Contains(pe.Error(), sentinel.URL) { + t.Fatalf("PollError renders the refused Location: %s", pe.Error()) + } + if sentinelHits.Load() != 0 { + t.Fatalf("the foreign origin received %d request(s), want zero egress", sentinelHits.Load()) + } + hooks.assertNoLeak(t, "leak") +} + +// TestLivePolls_ABare3xxIsRefusedToo: a 3xx with no Location is equally not +// a page: the seam reports redirect_refused with the fixed unparsable token. +func TestLivePolls_ABare3xxIsRefusedToo(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusFound) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused || pe.LocationOrigin != "unparsable" { + t.Fatalf("error = %v, want redirect_refused with the unparsable token", err) + } +} + +// TestLivePolls_ASameOriginRedirectIsRefusedToo: the API never redirects a +// feed call, and a continuation is followed by re-issuing the operation, so +// even a same-origin hop is answered by the guard — one request, no follow — +// and the seam reports the API origin as the refused one. +func TestLivePolls_ASameOriginRedirectIsRefusedToo(t *testing.T) { + var f *liveFixture + f = newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("position") == "pos-0" { + http.Redirect(w, r, f.server.URL+"/99999/events.json?position=pos-moved", http.StatusFound) + return + } + jsonResponse(w, 200, `{"events":[],"position":"pos-1"}`) + }) + _, err := f.live.Polls().Poll(context.Background(), eventfeed.Cursor{Position: "pos-0"}, eventfeed.Filters{}) + var pe *eventfeed.PollError + if !errors.As(err, &pe) || pe.Kind != eventfeed.PollRedirectRefused || pe.LocationOrigin != f.server.URL { + t.Fatalf("error = %v, want redirect_refused carrying %q", err, f.server.URL) + } + if f.requests.Load() != 1 { + t.Fatalf("requests = %d, want the one the hop was refused on", f.requests.Load()) + } +} + +// TestLiveMinter_ARedirectIsUnrecoverable: a mint that answers 3xx is out of +// contract and a fresh mint answers the same way; the guard strips the +// Location so nothing — the seam's error or the hooks' — renders it. +func TestLiveMinter_ARedirectIsUnrecoverable(t *testing.T) { + hooks := &recordingHooks{} + f := newLiveFixtureWith(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", "https://evil.example.test/mint?token=leak") + w.WriteHeader(http.StatusTemporaryRedirect) + }, basecamp.WithHooks(hooks)) + _, err := f.live.Minter().MintStreamTicket(context.Background()) + var me *eventfeed.MintError + if !errors.As(err, &me) || me.Kind != eventfeed.MintUnrecoverable { + t.Fatalf("error = %v, want MintUnrecoverable", err) + } + if strings.Contains(me.Error(), "leak") || strings.Contains(me.Error(), "evil") { + t.Fatalf("MintError renders the refused Location: %s", me.Error()) + } + hooks.assertNoLeak(t, "leak") +} + +func TestLivePolls_CancellationPassesThrough(t *testing.T) { + f := newLiveFixture(t, eventfeed.AccountLane, func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := f.live.Polls().Poll(ctx, eventfeed.Cursor{Since: "now"}, eventfeed.Filters{}) + done <- err + }() + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled passed through", err) + } + var pe *eventfeed.PollError + if errors.As(err, &pe) { + t.Fatalf("a cancelled call was classified as %s; the connector reads the context", pe.Kind) + } + case <-time.After(5 * time.Second): + t.Fatal("the cancelled poll did not return promptly") + } +} + +func TestNewLiveValidatesAndConnects(t *testing.T) { + cfg := basecamp.DefaultConfig() + cfg.BaseURL = "http://localhost:3001" + live, err := eventfeed.NewLive(cfg, &basecamp.StaticTokenProvider{Token: "t"}, "1", eventfeed.InboxLane) + if err != nil { + t.Fatalf("NewLive: %v", err) + } + if live.Origin() != "http://localhost:3001" || live.Client() == nil { + t.Fatalf("live = origin %q client %v", live.Origin(), live.Client()) + } + if _, err := live.Connect(eventfeed.WithFilters(eventfeed.Filters{Reasons: []string{"mentioned"}})); err != nil { + t.Fatalf("Connect on the inbox lane: %v", err) + } + if _, err := live.Connect(eventfeed.WithFilters(eventfeed.Filters{Creators: []int64{1}})); err == nil { + t.Fatal("Connect accepted creators on the inbox lane") + } + var te *eventfeed.TerminalError + if _, err := live.Connect(eventfeed.WithLane(eventfeed.AccountLane)); !errors.As(err, &te) || te.Reason != eventfeed.ReasonUsage { + t.Fatalf("Connect(WithLane(AccountLane)) on an inbox binding = %v, want a usage error, not a silent override", err) + } + if _, err := live.Connect(eventfeed.WithLane(eventfeed.InboxLane)); err != nil { + t.Fatalf("Connect(WithLane(InboxLane)) on an inbox binding: %v", err) + } + for name, tc := range map[string]struct { + cfg *basecamp.Config + id string + lane eventfeed.Lane + }{ + "nil config": {nil, "1", eventfeed.AccountLane}, + "unknown lane": {cfg, "1", eventfeed.Lane(9)}, + "cleartext base": {&basecamp.Config{BaseURL: "http://api.example.test"}, "1", eventfeed.AccountLane}, + "nonnumeric id": {cfg, "abc", eventfeed.AccountLane}, + "userinfo base": {&basecamp.Config{BaseURL: "https://user:s3cret-leak@api.example.test"}, "1", eventfeed.AccountLane}, + "invalid utf-8": {&basecamp.Config{BaseURL: "https://exa\xffmple.test"}, "1", eventfeed.AccountLane}, + "dot segments": {&basecamp.Config{BaseURL: "https://api.example.test/api/../v1"}, "1", eventfeed.AccountLane}, + "query": {&basecamp.Config{BaseURL: "https://api.example.test/api?x=s3cret-leak"}, "1", eventfeed.AccountLane}, + "fragment": {&basecamp.Config{BaseURL: "https://api.example.test/api#s3cret-leak"}, "1", eventfeed.AccountLane}, + "doubled slash": {&basecamp.Config{BaseURL: "https://api.example.test/api//v1"}, "1", eventfeed.AccountLane}, + "doubled root": {&basecamp.Config{BaseURL: "https://api.example.test//"}, "1", eventfeed.AccountLane}, + "no origin": {&basecamp.Config{BaseURL: "/s3cret-leak"}, "1", eventfeed.AccountLane}, + } { + t.Run(name, func(t *testing.T) { + _, err := eventfeed.NewLive(tc.cfg, &basecamp.StaticTokenProvider{Token: "t"}, tc.id, tc.lane) + var te *eventfeed.TerminalError + if !errors.As(err, &te) || te.Reason != eventfeed.ReasonUsage { + t.Fatalf("NewLive error = %v, want a usage-coded construction error", err) + } + if strings.Contains(err.Error(), "s3cret-leak") { + t.Fatalf("the construction error renders the base URL: %v", err) + } + }) + } +} diff --git a/go/pkg/basecamp/eventfeed/seams.go b/go/pkg/basecamp/eventfeed/seams.go index 6a90162dc..e75774aa2 100644 --- a/go/pkg/basecamp/eventfeed/seams.go +++ b/go/pkg/basecamp/eventfeed/seams.go @@ -244,7 +244,17 @@ type PollError struct { // offending list, verbatim. Msg string // LocationOrigin is the refused redirect Location reduced to its origin - // (redirect_refused only). DATA, never a rendering — the CloseError.Reason + // (redirect_refused only), or one of two fixed tokens in its place: + // `unparsable` when the Location was absent or yielded no complete + // origin (§9), and `unrecorded` when the hop was refused but the + // adapter could not attribute the origin to this call — a host hook + // that returns a context unrelated to the one it was handed drops the + // per-call record the transport writes it on. The refusal and its zero + // egress hold in every case; only the origin is lost. The two tokens + // stay distinct so a reader of `unparsable` can keep taking it for what + // it says — a statement about what the server sent. + // + // DATA, never a rendering — the CloseError.Reason // precedent: a hostile redirect can reflect the caller's bearer into a // host label, so no rendering may carry this value. PollError.Error // deliberately omits it, and a terminal built from a refused redirect diff --git a/ruby/lib/basecamp/http.rb b/ruby/lib/basecamp/http.rb index 5a4bc1aa5..db5dd2e70 100644 --- a/ruby/lib/basecamp/http.rb +++ b/ruby/lib/basecamp/http.rb @@ -629,7 +629,9 @@ def build_faraday_client def request(method, path, params: {}, body: nil, allow_cross_origin: false, operation: nil) url = build_url(path, allow_cross_origin: allow_cross_origin) - # Mutations don't retry on 429/5xx to avoid duplicating data + # Mutations don't retry on 429/5xx to avoid duplicating data. A mutation + # declared naturally idempotent (CreateStreamTicket) should retry per its + # operation metadata instead; routing it here is tracked in #901. if method == :get request_with_retry(method, url, params: params, allow_cross_origin: allow_cross_origin, operation: operation) else diff --git a/spec/api-gaps/event-feed.md b/spec/api-gaps/event-feed.md index 6e755a332..ee12dde92 100644 --- a/spec/api-gaps/event-feed.md +++ b/spec/api-gaps/event-feed.md @@ -48,17 +48,23 @@ templates, templatifications, backlinks, bulk enrollments and the unscoped recording show, each needing its own triage); the three routes are waived in `spec/bc3-route-allowlist.yml` with the evidence, to be deleted at the repin. -What merged differs from the pre-merge record below in four places the -connector (SPEC §23, layer 2) still has to absorb: the digest scheme is -published as **`srv2`** (the `reasons` dimension joined it); the feed's 410 -`resume` re-enters at **`since=`**, not `since=now`, while the -inbox's re-enters at `since=0`; the filter set grew `performers`, -`exclude_performers` (both accepting the literal `self`) and `actor_types`; and -the inbox is a lane of its own (`GET /inbox.json`, agents only, items keyed by -`addressing_id`, 30-day retention). Layer 2 — the connector, the -`conformance/event-feed/` fixture family and its gate — continues on its own -branches against SPEC §23, which keeps its provisional markings until it -re-verifies against the merged head. +What merged differed from the pre-merge record below in four places, and the +connector (SPEC §23, layer 2) absorbed all four: the digest scheme is published +as **`srv2`** (the `reasons` dimension joined it) and the connector's checkpoint +namespace and 409 handling follow it; the feed's 410 `resume` re-enters at +**`since=`**, not `since=now`, while the inbox's re-enters at +`since=0`, and the connector treats the accepted resume as a position-resume +entry; the filter set grew `performers`, `exclude_performers` (both accepting +the literal `self`) and `actor_types`, which the connector validates +syntactically and carries into the digest; and the inbox is a lane of its own +(`GET /inbox.json`, agents only, items keyed by `addressing_id`, 30-day +retention), which the connector runs as `InboxLane` with a lane-keyed +checkpoint. The Go adapters (`eventfeed.NewLive`) bind the seams to the +generated `PollEvents`, `PollInbox` and `CreateStreamTicket` operations, and the +`conformance/event-feed/` fixture family and its gate pin the absorbed contract. +One wire-layer residue remains: Ruby's `create_stream_ticket` runs through the +runtime's single-attempt mutation path rather than the operation's declared +retry budget (#901). The pre-merge record follows, unedited, as history. diff --git a/spec/tracking-issues.yml b/spec/tracking-issues.yml index 94b15b08a..cbb0e762a 100644 --- a/spec/tracking-issues.yml +++ b/spec/tracking-issues.yml @@ -35,6 +35,18 @@ # sentence promising an issue owns something. Decrement or remove when the # sentence goes, and the gate stops asking. prose_tracking_issues: + - issue: 915 + file: "go/pkg/basecamp/eventfeed/live.go" + site: "go/pkg/basecamp/eventfeed/live.go — the 409 digest and 400 reason shapes are held at the adapter until the wrapper's decode holds them" + mentions: 1 + - issue: 915 + file: "go/pkg/basecamp/eventfeed/live_test.go" + site: "go/pkg/basecamp/eventfeed/live_test.go — the inbox 410 accepts a member its shape does not declare until the wrapper's decode refuses one" + mentions: 1 + - issue: 901 + file: "ruby/lib/basecamp/http.rb" + site: "ruby/lib/basecamp/http.rb — declared-idempotent mutations such as CreateStreamTicket bypass the retry budget" + mentions: 1 - issue: 589 file: "Makefile" site: "Makefile — the vendored bc3-route table's freshness gate needs BC3_REPO_PATH, so it is not in CI" @@ -119,19 +131,3 @@ prose_tracking_issues: file: "SPEC.md" site: "SPEC.md Appendix F — OAuth endpoint address enforcement beyond Go (§16 req 5–6) is pending #818's umbrella" mentions: 2 - - issue: 819 - file: "SPEC.md" - site: "SPEC.md §23 zero-egress paragraph — the Layer-1 302 test's adapters are pending #819" - mentions: 1 - - issue: 819 - file: "conformance/event-feed/README.md" - site: "conformance/event-feed/README.md — Layer-1 seam-adapter conformance (rows 15/30 and the row-15 note) is pending #819" - mentions: 3 - - issue: 819 - file: "conformance/event-feed/fixtures/30-continuation-redirect-cross-origin.json" - site: "fixture 30's description — the below-seam zero-request proof is pending #819's Layer-1 adapters" - mentions: 1 - - issue: 819 - file: "conformance/event-feed/schema.json" - site: "schema.json fixture-30 respond variant — the Layer-1 302 test is pending #819" - mentions: 1