// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial // SPDX-FileCopyrightText: 2025-2026 Christian Galo package server import ( "context" "database/sql" "errors" "fmt" "log/slog" "net/http" "time" "git.coopcloud.tech/wiki-cafe/member-console/internal/instance" "git.coopcloud.tech/wiki-cafe/member-console/internal/integration" stripedb "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store" stripewf "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues" "go.temporal.io/sdk/temporal" ) // The Stripe provider page (design D24; integration-settings ADDED // requirement "The Stripe provider page carries its status and the // delivery-queue report"): Stripe's own home, the destination of its name // on the Integrations table and the overview's Integrations card, distinct // from its settings page (which now carries configuration only). Carries // the configured state, the mode, and the Delivery queue section moved // here from the settings page (operator_integration_settings.go). // StripeIntegrationData is the body for operator_integration_stripe.html. type StripeIntegrationData struct { // Configured/Missing: the same required-key resolution // (configurationReadiness) the settings page and the Integrations // table use, so all three surfaces read the same fact. Configured bool Missing []string // ModeLabel is "Test mode" or "Live mode", from the mode derived from // the Stripe API key at boot // โ€” mirrors OverviewStripeFacts.ModeLabel (operator_overview.go) so the // overview's Stripe row and this page never disagree. ModeLabel string // Description is the provider manifest's declared description, when it // carries one ("" otherwise); Header() falls back to a generic lead. Description string // DeliveryQueueHeader/DeliveryQueue/DeliveryQueueEntries carry the // "Delivery queue" section, moved here from the settings page // (integration-settings ADDED requirement: "The Stripe provider page // carries its status and the delivery-queue report"). Also read by the // landing surface's Stripe element (operator_overview.go's // loadOverviewStripeFacts, which only summarizes DeliveryQueue) so the // two surfaces can never disagree about what the outbox holds. DeliveryQueueHeader SectionHeader DeliveryQueue DeliveryQueue DeliveryQueueEntries []DeadLetterEntry // InboundEventsHeader/InboundEvents/InboundEventEntries carry the // "Inbound events" section: Stripe's webhook events that failed // processing through every retry round and were dead-lettered // (stripe-integration-infrastructure: "Failed events are retried, then // dead-lettered"). The outbox section above is outbound work; this is // the inbound counterpart, and the overview's Stripe row counts both. InboundEventsHeader SectionHeader InboundEvents InboundEvents InboundEventEntries []DeadLetterEntry // EnvironmentCheckHeader/EnvironmentCheck carry the "Environment // check" section (stripe-environment-stamp D3): where the read-back of // the mapped Stripe ids stands under the key in force, and the control // that runs it again. Rendered only when Stripe is configured, since // there is no key to read anything back under otherwise. EnvironmentCheckHeader SectionHeader EnvironmentCheck EnvironmentCheck } // EnvironmentCheck is the Environment check section's state: one resting // line naming where the read-back stands, and the control beside it // (stripe-environment-stamp D3). type EnvironmentCheck struct { Line string } // InboundEvents is the health of Stripe's inbound webhook events: how many // have exhausted their retries, and how many arrived from the environment // the configured key cannot reach. Received, retrying and processing rows // need no operator and are not counted. type InboundEvents struct { DeadLetter int64 // Refused counts the events captured from the other Stripe environment // (stripe-environment-stamp D6). They are not a failure an operator // retries: nothing here will ever process them, and the count is how a // deployment learns an old endpoint is still pointed at it. Refused int64 // RefusedLine names where those events came from and which key they // met. Empty when none has been refused, so the section says nothing // about an environment nothing arrived from. RefusedLine string Available bool } // NeedsAttention reports whether inbound events hold work that will not // resolve on its own. func (e InboundEvents) NeedsAttention() bool { return e.DeadLetter > 0 } // Quiet reports whether the section has nothing to report: no event has // exhausted its retries and none has arrived from the other environment. func (e InboundEvents) Quiet() bool { return e.DeadLetter == 0 && e.Refused == 0 } // loadInboundEvents counts Stripe's dead-lettered and refused webhook // events. Raw SQL against core.webhook_events, as loadDeadLetterEntries // does for the outbox; a nil db or a failed probe leaves Available false. // keyMode is the environment the configured key is in, which is what makes // the refused line sayable: with two environments, an event the key // refused came from the other one. func loadInboundEvents(ctx context.Context, db *sql.DB, keyMode string, logger *slog.Logger) InboundEvents { if db == nil { return InboundEvents{} } var deadLetter, refused int64 if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FILTER (WHERE status = 'dead_letter'), COUNT(*) FILTER (WHERE status = 'refused') FROM core.webhook_events WHERE provider = 'stripe'`, ).Scan(&deadLetter, &refused); err != nil { logger.Warn("inbound events: status count failed", slog.Any("error", err)) return InboundEvents{} } return InboundEvents{ DeadLetter: deadLetter, Refused: refused, RefusedLine: refusedEventsLine(refused, keyMode), Available: true, } } // refusedEventsLine states the refused events in the two environments' // own words: the one they came from and the one the key is in. No key // configured refuses nothing, so there is no line to write. func refusedEventsLine(refused int64, keyMode string) string { if refused == 0 || keyMode == "" { return "" } eventMode := stripedb.ModeTest if keyMode == stripedb.ModeTest { eventMode = stripedb.ModeLive } return fmt.Sprintf("%d %s arrived in %s mode under a %s key.", refused, pluralize(refused, "event", "events"), eventMode, keyMode) } // refusedEventDetail is what a refused row says in the table's detail // column: it carries no error, because nothing failed. The event came from // the environment the key cannot read, and was kept rather than processed // (stripe-environment-stamp D6). const refusedEventDetail = "Refused, mode mismatch" // loadInboundDeadLetterEntries lists Stripe's dead-lettered and refused // webhook events for the Inbound events table, capped like the outbox // table. OperationType carries the event type (e.g. "invoice.paid"). // // Dead-lettered rows sort ahead of refused ones, and each group is newest // first. The two statuses share a table but not a reader's stake: a // dead-lettered event needs an operator, while a refused one is recorded // and finished with. An old endpoint still delivering to a moved key can // refuse events by the hundred, and under one time ordering that flood // pushes every dead-lettered row past the cap. func loadInboundDeadLetterEntries(ctx context.Context, db *sql.DB, logger *slog.Logger) []DeadLetterEntry { if db == nil { return nil } rows, err := db.QueryContext(ctx, `SELECT event_type, COALESCE(error_message, ''), retry_count, updated_at, status FROM core.webhook_events WHERE provider = 'stripe' AND status IN ('dead_letter', 'refused') ORDER BY (status = 'dead_letter') DESC, updated_at DESC LIMIT $1`, deadLetterEntryLimit) if err != nil { logger.Warn("inbound events: dead-letter entries query failed", slog.Any("error", err)) return nil } defer rows.Close() var entries []DeadLetterEntry for rows.Next() { var e DeadLetterEntry var updatedAt time.Time var status string if err := rows.Scan(&e.OperationType, &e.ErrorMessage, &e.Attempts, &updatedAt, &status); err != nil { logger.Warn("inbound events: dead-letter entry scan failed", slog.Any("error", err)) return entries } if status == "refused" { e.Refused = true e.ErrorMessage = refusedEventDetail } e.UpdatedAt = updatedAt.Format("Jan 2, 2006 3:04 PM") entries = append(entries, e) } if err := rows.Err(); err != nil { logger.Warn("inbound events: dead-letter entries rows failed", slog.Any("error", err)) } return entries } // DeliveryQueue is the shared integration outbox's health, split by what an // operator would do about each bucket. Retrying rows are in-flight and need // no action; DeadLetter rows have exhausted their retries and will not move // without intervention, so only that bucket gets alarm styling // (integration-settings: "Only dead-letter outbox work raises alarm // styling"). type DeliveryQueue struct { Pending int64 Retrying int64 DeadLetter int64 Available bool } // NeedsAttention reports whether the delivery queue holds work that will // not resolve on its own. func (q DeliveryQueue) NeedsAttention() bool { return q.DeadLetter > 0 } // Queued is the draining-normally count: enqueued or retrying work, as // opposed to the dead-lettered work NeedsAttention names. func (q DeliveryQueue) Queued() int64 { return q.Pending + q.Retrying } // loadDeliveryQueue probes the shared integration outbox (core.outbox). // Today only Stripe enqueues into it (operator_billing.go); an unavailable // probe leaves Available false rather than a set of zeroes neither caller // could stand behind. A nil querier (a handler built without one, as some // tests do) also degrades to unavailable rather than panicking. func loadDeliveryQueue(ctx context.Context, iq integration.Querier, logger *slog.Logger) DeliveryQueue { if iq == nil { return DeliveryQueue{} } counts, err := iq.CountOutboxByStatus(ctx) if err != nil { logger.Warn("delivery queue: outbox health probe failed", slog.Any("error", err)) return DeliveryQueue{} } return DeliveryQueue{ Pending: counts.PendingCount, Retrying: counts.FailedCount, DeadLetter: counts.DeadLetterCount, Available: true, } } // DeadLetterEntry is one dead-lettered outbox row: it has exhausted its // retries and will not move without an operator (integration-settings: // "the dead-letter table with its operation identifiers rendered as // "). OperationType is the outbox's action_type column (e.g. // "create_stripe_price") โ€” the write the queue was attempting. type DeadLetterEntry struct { OperationType string ErrorMessage string Attempts int32 UpdatedAt string // Refused marks a row the table lists for a reason other than // exhausted retries: an inbound event captured from the other Stripe // environment. It carries no alarm styling, because no operator can // move it and none should try (stripe-environment-stamp D6). Outbox // entries never set it. Refused bool } // deadLetterEntryLimit caps the Delivery queue section's table so a // long-neglected queue cannot blow up the provider page; the counts above // (DeliveryQueue.DeadLetter) still state the true total regardless. const deadLetterEntryLimit = 50 // loadDeadLetterEntries lists the Stripe outbox's dead-lettered rows for // the Delivery queue section's table. Raw SQL against core.outbox directly // (the same cross-schema pattern operator_pages.go's ownedResourceKeys // uses) rather than a new query on the integration module, which this lane // does not own. Degrades to an empty list on failure โ€” the counts above // still say a dead-letter bucket exists even when the row-level detail // cannot be fetched, and a nil db (a handler built without one, as some // tests do) degrades the same way rather than panicking. func loadDeadLetterEntries(ctx context.Context, db *sql.DB, logger *slog.Logger) []DeadLetterEntry { if db == nil { return nil } rows, err := db.QueryContext(ctx, `SELECT action_type, COALESCE(error_message, ''), attempts, updated_at FROM core.outbox WHERE provider = 'stripe' AND status = 'dead_letter' ORDER BY updated_at DESC LIMIT $1`, deadLetterEntryLimit) if err != nil { logger.Warn("delivery queue: dead-letter entries query failed", slog.Any("error", err)) return nil } defer rows.Close() var entries []DeadLetterEntry for rows.Next() { var e DeadLetterEntry var updatedAt time.Time if err := rows.Scan(&e.OperationType, &e.ErrorMessage, &e.Attempts, &updatedAt); err != nil { logger.Warn("delivery queue: dead-letter entry scan failed", slog.Any("error", err)) return entries } e.UpdatedAt = updatedAt.Format("Jan 2, 2006 3:04 PM") entries = append(entries, e) } if err := rows.Err(); err != nil { logger.Warn("delivery queue: dead-letter entries rows failed", slog.Any("error", err)) } return entries } // environmentCheckDateFormat is the page's existing date format, shared // with the two dead-letter tables above. const environmentCheckDateFormat = "Jan 2, 2006 3:04 PM" // loadEnvironmentCheck states where the read-back of the mapped Stripe ids // stands, in one line (stripe-environment-stamp D3). Four states, each // with its own fact to carry: // // - no record, and nothing running: the check has never run here; // - a record with a start and no finish: a run is under way, and the // line says how many ids it is reading back; // - a finished record under the key in force: when it ran and what it // found; // - a finished record under another key: the key moved since, so the // counts describe ids nobody has confirmed under this one, and the // line says how many are waiting. // // fingerprint is the digest of the key this process holds. A failed read // degrades to the never-run line rather than to nothing: the section still // offers the control, which is the way out of any of these states. func loadEnvironmentCheck(ctx context.Context, db *sql.DB, fingerprint string, logger *slog.Logger) EnvironmentCheck { if db == nil { return EnvironmentCheck{Line: environmentNeverChecked} } raw, found, err := instance.NewStore(db).GetJSON(ctx, instance.StripeEnvironmentCheck) if err != nil { logger.Warn("environment check: record read failed", slog.Any("error", err)) return EnvironmentCheck{Line: environmentNeverChecked} } rec, ok := stripedb.DecodeEnvironmentCheckRecord(raw, found) if !ok || rec.StartedAt == nil { return EnvironmentCheck{Line: environmentNeverChecked} } // Only the state that needs a count pays for one. var mapped int64 switch { case rec.Running(): mapped = countMappedStripeIDs(ctx, db, false, logger) case rec.KeyFingerprint != fingerprint: mapped = countMappedStripeIDs(ctx, db, true, logger) } return EnvironmentCheck{Line: environmentCheckLine(rec, fingerprint, mapped)} } // environmentNeverChecked is the line for a deployment where the read-back // has not run: no record, an unreadable one, or one written before a run // ever started. const environmentNeverChecked = "Not checked yet." // environmentCheckLine writes the resting line for a record that exists. // mapped is the count the state needs: the ids a run in progress is // reading back, or the ids left unverified after the key moved. The // states that need no count ignore it. func environmentCheckLine(rec stripedb.EnvironmentCheckRecord, fingerprint string, mapped int64) string { switch { case rec.StartedAt == nil: return environmentNeverChecked case rec.Running(): return fmt.Sprintf("Checking %d Stripe %s under the current key.", mapped, pluralize(mapped, "id", "ids")) case rec.KeyFingerprint != fingerprint: return fmt.Sprintf("API key changed since the last check. %d %s unverified.", mapped, pluralize(mapped, "mapping", "mappings")) } return fmt.Sprintf("Last checked %s. %d verified, %d stale.", rec.FinishedAt.Local().Format(environmentCheckDateFormat), rec.Verified(), rec.Stale) } // countMappedStripeIDs counts the product and price mappings holding a // Stripe id, or, with unverifiedOnly, the subset no check has confirmed // under the key in force. Raw SQL across the stripe schema, the way // loadInboundEvents reads core.webhook_events: the two counts exist for // one line of copy each and are not worth a generated query. A failed // count reads zero, which makes the line say less rather than wrong. func countMappedStripeIDs(ctx context.Context, db *sql.DB, unverifiedOnly bool, logger *slog.Logger) int64 { query := `SELECT (SELECT COUNT(*) FROM stripe.product_mappings WHERE stripe_product_id IS NOT NULL) + (SELECT COUNT(*) FROM stripe.price_mappings WHERE stripe_price_id IS NOT NULL)` if unverifiedOnly { query = `SELECT (SELECT COUNT(*) FROM stripe.product_mappings WHERE stripe_product_id IS NOT NULL AND verified_at IS NULL) + (SELECT COUNT(*) FROM stripe.price_mappings WHERE stripe_price_id IS NOT NULL AND verified_at IS NULL)` } var n int64 if err := db.QueryRowContext(ctx, query).Scan(&n); err != nil { logger.Warn("environment check: mapping count failed", slog.Any("error", err)) return 0 } return n } // stripeModeLabel names the mode derived from the Stripe API key at boot // (server.Config.StripeMode). A deployment with no key configured reads // "Test mode": nothing it shows is live money, and the page states // separately that Stripe is unconfigured. func stripeModeLabel(mode string) string { if mode == "live" { return "Live mode" } return "Test mode" } // GetStripeIntegrationPage handles GET /operator/integrations/stripe. func (h *OperatorPartialsHandler) GetStripeIntegrationPage(w http.ResponseWriter, r *http.Request) { configured, missing := configurationReadiness(h.IntegrationConfigs, "stripe") modeLabel := stripeModeLabel(h.StripeMode) bodyData := StripeIntegrationData{ Configured: configured, Missing: missing, ModeLabel: modeLabel, Description: stripedb.ProviderSource().ProviderManifest().Description, EnvironmentCheckHeader: SectionHeader{Title: "Environment check"}, DeliveryQueueHeader: SectionHeader{Title: "Delivery queue"}, InboundEventsHeader: SectionHeader{Title: "Inbound events"}, } var iq integration.Querier if h.Database != nil { iq = integration.New(h.Database) } bodyData.EnvironmentCheck = loadEnvironmentCheck(r.Context(), h.Database, h.StripeKeyFingerprint, h.Logger) bodyData.DeliveryQueue = loadDeliveryQueue(r.Context(), iq, h.Logger) bodyData.DeliveryQueueEntries = loadDeadLetterEntries(r.Context(), h.Database, h.Logger) bodyData.InboundEvents = loadInboundEvents(r.Context(), h.Database, h.StripeMode, h.Logger) bodyData.InboundEventEntries = loadInboundDeadLetterEntries(r.Context(), h.Database, h.Logger) page := h.buildOperatorPageData(r) page.IAPosition = "integration:integrations:stripe" page.ActiveCapability = "integrations" page.BodyTemplate = "operator_integration_stripe.html" page.BodyData = bodyData h.Templates.Render(w, "operator.html", page) } // environmentCheckWait bounds how long the control's request waits on the // run before answering. A catalog of a few dozen ids settles inside it, so // the common answer is the counts; a larger one answers that the check // started and the section says the rest on the next load. The wait is // short because it is a request holding a worker, not a job. const environmentCheckWait = 3 * time.Second // PostStripeEnvironmentCheck handles // POST /partials/operator/integrations/stripe/environment-check: the // provider page's "Check now" control (stripe-environment-stamp // D3). It starts the same workflow boot starts, under the same id, so an // operator pressing it while the boot-started run is still going is // refused by Temporal and told so. // // It waits a few seconds on the result rather than none: an operator who // pressed the control wants the outcome, and a catalog this size usually // has one by then. The section re-renders either way, so the page never // disagrees with the record. func (h *OperatorPartialsHandler) PostStripeEnvironmentCheck(w http.ResponseWriter, r *http.Request) { if !h.StripeConfigured { http.Error(w, "Stripe is not configured.", http.StatusNotFound) return } if h.TemporalClient == nil { fireErrorToast(w, "Temporal is unavailable.") h.renderStripeEnvironmentCheck(w, r) return } run, err := stripewf.StartEnvironmentCheck(r.Context(), h.TemporalClient, queues.Main, stripewf.EnvironmentCheckTriggerOperator) if err != nil { if temporal.IsWorkflowExecutionAlreadyStartedError(err) { fireSuccessToast(w, "Check running.") h.renderStripeEnvironmentCheck(w, r) return } h.Logger.Error("failed to start the stripe environment check workflow", slog.Any("error", err)) fireErrorToast(w, "Could not start the check.") h.renderStripeEnvironmentCheck(w, r) return } waitCtx, cancel := context.WithTimeout(r.Context(), environmentCheckWait) defer cancel() var result stripewf.EnvironmentCheckResult // Get returns an error for two different facts: the wait ran out while // the run went on, and the run itself failed. Only the deadline means // the check is still going, so the two are told apart by the wait // context rather than reported as one. switch err := run.Get(waitCtx, &result); { case err == nil: fireSuccessToast(w, fmt.Sprintf("%d checked, %d stale.", result.Checked, result.Stale)) case errors.Is(waitCtx.Err(), context.DeadlineExceeded): fireSuccessToast(w, "Check started.") default: // The failure itself goes to the log, not the toast: a workflow // error carries the activity's wrapped Go error, which is the // leak docs/operator-ux-conventions.md ยง4 keeps out of the UI. h.Logger.Error("the stripe environment check failed", slog.Any("error", err)) fireErrorToast(w, "The check failed.") } h.renderStripeEnvironmentCheck(w, r) } // renderStripeEnvironmentCheck re-renders the Environment check section // alone, which is the whole of what the control changes. func (h *OperatorPartialsHandler) renderStripeEnvironmentCheck(w http.ResponseWriter, r *http.Request) { h.Templates.Render(w, "operator_integration_stripe_environment.html", loadEnvironmentCheck(r.Context(), h.Database, h.StripeKeyFingerprint, h.Logger)) }