diff --git a/.env.example b/.env.example index 86c22df3..b8b2eb60 100644 --- a/.env.example +++ b/.env.example @@ -121,6 +121,19 @@ INGEST_REQUIRE_TLS=false # HAZEL_OAUTH_CLIENT_SECRET= # HAZEL_OAUTH_SCOPES=openid email profile organizations:read channels:read channel-webhooks:write +# Slack integration (bot install via OAuth v2) +# Maple installs a Slack app into a workspace via OAuth. Create a Slack app at +# https://api.slack.com/apps, add the bot scopes, and set the redirect URL to +# /oauth/slack/callback. Both values are optional — the integration +# stays disabled until they are set. +# SLACK_CLIENT_ID= +# SLACK_CLIENT_SECRET= +# Optional dedicated bearer secret for the internal Slack-bot resolve endpoint +# (`GET /internal/slack/workspaces/:teamId`), so the Railway-hosted Slack agent +# can be given a secret distinct from INTERNAL_SERVICE_TOKEN. Falls back to +# INTERNAL_SERVICE_TOKEN when unset. +# SLACK_INTERNAL_SERVICE_TOKEN= + # GitHub integration (GitHub App) # Maple connects to GitHub via a GitHub App for repo access, OAuth sign-in, and # webhooks. Create one at https://github.com/settings/apps (or your org's diff --git a/.gitignore b/.gitignore index 191bcb60..8aee6c59 100644 --- a/.gitignore +++ b/.gitignore @@ -32,6 +32,8 @@ apps/api/.data/ apps/api/scripts/.bench/ .alchemy +# eve agent build/model-catalog cache +.eve .mcp.json .dev.vars apps/chat-agent/.dev.vars diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index 75b954ff..fcac5a5f 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -207,6 +207,9 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => ...optionalPlain("HAZEL_OAUTH_CLIENT_ID"), ...optionalSecret("HAZEL_OAUTH_CLIENT_SECRET"), ...optionalPlain("HAZEL_OAUTH_SCOPES"), + // Slack integration (bot install via OAuth v2) + ...optionalPlain("SLACK_CLIENT_ID"), + ...optionalSecret("SLACK_CLIENT_SECRET"), ...optionalPlain("GITHUB_APP_ID"), ...optionalPlain("GITHUB_APP_SLUG"), ...optionalSecret("GITHUB_APP_PRIVATE_KEY"), diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 3484e78e..5d752cfe 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -17,6 +17,7 @@ import { HttpV2ApiKeysLive } from "./routes/v2/api-keys.http" import { HttpV2AttributeMappingsLive } from "./routes/v2/attribute-mappings.http" import { HttpV2DashboardsLive } from "./routes/v2/dashboards.http" import { HttpV2IngestKeysLive } from "./routes/v2/ingest-keys.http" +import { HttpV2SlackIntegrationsLive } from "./routes/v2/integrations.http" import { HttpV2ErrorIssuesLive } from "./routes/v2/error-issues.http" import { HttpV2AnomaliesLive } from "./routes/v2/anomalies.http" import { HttpV2InvestigationsLive } from "./routes/v2/investigations.http" @@ -47,6 +48,7 @@ import { OAuthDiscoveryRouter } from "./routes/oauth-discovery.http" import { HttpOrgClickHouseSettingsLive } from "./routes/org-clickhouse-settings.http" import { HttpOrganizationsLive } from "./routes/organizations.http" import { PlanetScaleWebhookRouter } from "./routes/planetscale-webhook.http" +import { SlackCallbackRouter, SlackInternalRouter } from "./routes/slack-integration.http" import { PrometheusScrapeProxyRouter } from "./routes/prometheus-scrape-proxy.http" import { ScraperInternalRouter } from "./routes/scraper-internal.http" import { VcsWebhookRouter } from "./routes/vcs-webhook.http" @@ -92,6 +94,7 @@ import { PlanetScaleDiscoveryService } from "./services/PlanetScaleDiscoveryServ import { PlanetScaleOAuthService } from "./services/PlanetScaleOAuthService" import { PlanetScaleService } from "./services/PlanetScaleService" import { ScrapeTargetsService } from "./services/ScrapeTargetsService" +import { SlackIntegrationService } from "./services/SlackIntegrationService" import { WarehouseQueryService } from "./lib/WarehouseQueryService" import { OAuthStateRepository } from "./services/OAuthStateRepository" import { GithubAppClient } from "./services/vcs/vendor/github/GithubAppClient" @@ -205,6 +208,13 @@ const NotificationDispatcherLive = NotificationDispatcher.layer.pipe( Layer.provideMerge(Layer.mergeAll(CoreServicesLive, EmailServiceLive)), ) +// Slack integration: OAuth install/callback, status, channels, uninstall, and +// the internal bot-resolve endpoint. Needs ApiKeysService (mint the bot key) + +// OAuthStateRepository (CSRF state) on top of the core services. +const SlackIntegrationServiceLive = SlackIntegrationService.layer.pipe( + Layer.provideMerge(Layer.mergeAll(CoreServicesLive, OAuthStateRepository.layer)), +) + const ErrorsServiceLive = ErrorsService.layer.pipe( Layer.provideMerge( Layer.mergeAll( @@ -272,6 +282,7 @@ export const MainLive = Layer.mergeAll( DigestServiceLive, DemoServiceLive, VcsServicesLive, + SlackIntegrationServiceLive, ) const ApiRoutes = HttpApiBuilder.layer(MapleApi).pipe( @@ -312,6 +323,7 @@ const ApiV2Routes = HttpApiBuilder.layer(MapleApiV2).pipe( HttpV2AlertDestinationsLive, HttpV2AlertIncidentsLive, HttpV2IngestKeysLive, + HttpV2SlackIntegrationsLive, HttpV2ErrorIssuesLive, HttpV2AttributeMappingsLive, HttpV2ScrapeTargetsLive, @@ -334,6 +346,8 @@ export const AllRoutes = Layer.mergeAll( ApiRoutes, ApiV2Routes, IntegrationsCallbackRouter, + SlackCallbackRouter, + SlackInternalRouter, OAuthDiscoveryRouter, PlanetScaleWebhookRouter, PrometheusScrapeProxyRouter, diff --git a/apps/api/src/lib/Env.ts b/apps/api/src/lib/Env.ts index 26052e0b..c7338bdc 100644 --- a/apps/api/src/lib/Env.ts +++ b/apps/api/src/lib/Env.ts @@ -44,6 +44,14 @@ export interface EnvShape { readonly HAZEL_OAUTH_CLIENT_ID: Option.Option readonly HAZEL_OAUTH_CLIENT_SECRET: Option.Option> readonly HAZEL_OAUTH_SCOPES: string + readonly SLACK_CLIENT_ID: Option.Option + readonly SLACK_CLIENT_SECRET: Option.Option> + /** + * Optional dedicated bearer secret for the internal Slack-bot resolve + * endpoint, so the Railway bot can hold a token distinct from the + * MCP-internal `INTERNAL_SERVICE_TOKEN`. Falls back to that token when unset. + */ + readonly SLACK_INTERNAL_SERVICE_TOKEN: Option.Option> readonly GITHUB_APP_ID: Option.Option readonly GITHUB_APP_SLUG: Option.Option readonly GITHUB_APP_PRIVATE_KEY: Option.Option> @@ -126,6 +134,9 @@ const envConfig = Config.all({ "HAZEL_OAUTH_SCOPES", "openid email profile organizations:read channels:read channel-webhooks:write", ), + SLACK_CLIENT_ID: optionalString("SLACK_CLIENT_ID"), + SLACK_CLIENT_SECRET: optionalRedacted("SLACK_CLIENT_SECRET"), + SLACK_INTERNAL_SERVICE_TOKEN: optionalRedacted("SLACK_INTERNAL_SERVICE_TOKEN"), GITHUB_APP_ID: optionalString("GITHUB_APP_ID"), GITHUB_APP_SLUG: optionalString("GITHUB_APP_SLUG"), GITHUB_APP_PRIVATE_KEY: optionalRedacted("GITHUB_APP_PRIVATE_KEY"), diff --git a/apps/api/src/mcp/dispatcher.test.ts b/apps/api/src/mcp/dispatcher.test.ts index e7eaad0b..922bf78c 100644 --- a/apps/api/src/mcp/dispatcher.test.ts +++ b/apps/api/src/mcp/dispatcher.test.ts @@ -18,6 +18,13 @@ describe("MCP dispatcher", () => { }), ) + it("emits an object-typed inputSchema for every tool (strict MCP clients require it)", () => { + for (const definition of mapleToolDefinitions) { + const inputSchema = toInputSchema(definition.schema) + expect(inputSchema.type, `${definition.name} inputSchema.type`).toBe("object") + } + }) + it.effect("returns MCP validation feedback for invalid model tool input", () => Effect.gen(function* () { const result = yield* callMcpTool("inspect_trace", {}) as unknown as Effect.Effect< diff --git a/apps/api/src/mcp/tools/registry.ts b/apps/api/src/mcp/tools/registry.ts index 8563953b..18c870c7 100644 --- a/apps/api/src/mcp/tools/registry.ts +++ b/apps/api/src/mcp/tools/registry.ts @@ -69,9 +69,25 @@ export interface MapleToolDefinition { export const toInputSchema = (schema: Schema.Top): Record => { const document = Schema.toJsonSchemaDocument(schema) - return Object.keys(document.definitions).length > 0 - ? { ...document.schema, $defs: document.definitions } - : document.schema + const base = + Object.keys(document.definitions).length > 0 + ? { ...document.schema, $defs: document.definitions } + : document.schema + // MCP requires the top-level inputSchema to be an object schema (`type: "object"`). + // Effect emits `{ anyOf: [{ type: "object" }, { type: "array" }] }` for an empty + // `Struct({})` (a no-parameter tool), which strict MCP clients reject — the Vercel + // AI SDK's `tools/list` Zod validator fails on `inputSchema.type` and drops EVERY + // tool from the connection. Normalize a no-property struct to an explicit empty + // object schema. `$ref` roots (hoisted schemas) already carry a valid object type. + if (base.type !== "object" && !("$ref" in base)) { + return { + type: "object", + properties: {}, + additionalProperties: false, + ...("$defs" in base ? { $defs: base.$defs } : {}), + } + } + return base } const collectMapleToolDefinitions = (): ReadonlyArray => { diff --git a/apps/api/src/routes/slack-integration.http.test.ts b/apps/api/src/routes/slack-integration.http.test.ts new file mode 100644 index 00000000..cd902a6e --- /dev/null +++ b/apps/api/src/routes/slack-integration.http.test.ts @@ -0,0 +1,263 @@ +import { createCipheriv, randomBytes } from "node:crypto" +import { afterEach, assert, describe, it } from "@effect/vitest" +import { ConfigProvider, Effect, Layer } from "effect" +import { HttpRouter } from "effect/unstable/http" +import { Env } from "../lib/Env" +import { cleanupTestDbs, createTestDb, executeSql, type TestDb } from "../lib/test-pglite" +import { ApiKeysService } from "../services/ApiKeysService" +import { OAuthStateRepository } from "../services/OAuthStateRepository" +import { SlackIntegrationService } from "../services/SlackIntegrationService" +import { SlackInternalRouter } from "./slack-integration.http" + +const trackedDbs: TestDb[] = [] +afterEach(() => cleanupTestDbs(trackedDbs)) + +const ENCRYPTION_KEY = Buffer.alloc(32, 7) + +/** AES-256-GCM encrypt matching Crypto.ts's format (12-byte iv, base64 fields). */ +const encryptField = (plaintext: string, key: Buffer) => { + const iv = randomBytes(12) + const cipher = createCipheriv("aes-256-gcm", key, iv) + const ciphertext = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()]) + return { + ciphertext: ciphertext.toString("base64"), + iv: iv.toString("base64"), + tag: cipher.getAuthTag().toString("base64"), + } +} + +/** Insert an encrypted slack_workspaces row directly (bypasses OAuth). */ +const insertWorkspace = async ( + testDb: TestDb, + opts: { + id: string + orgId: string + teamId: string + teamName: string + botToken: string + apiKey: string + revoked?: boolean + }, +) => { + const bot = encryptField(opts.botToken, ENCRYPTION_KEY) + const key = encryptField(opts.apiKey, ENCRYPTION_KEY) + await executeSql( + testDb, + `INSERT INTO slack_workspaces ( + id, org_id, team_id, team_name, bot_user_id, scope, + bot_token_ciphertext, bot_token_iv, bot_token_tag, + api_key_id, api_key_secret_ciphertext, api_key_secret_iv, api_key_secret_tag, + installed_by_user_id, created_at, updated_at, revoked_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14, now(), now(), ${opts.revoked ? "now()" : "NULL"})`, + [ + opts.id, + opts.orgId, + opts.teamId, + opts.teamName, + "U0BOT", + "chat:write", + bot.ciphertext, + bot.iv, + bot.tag, + "11111111-2222-4333-8444-555555555555", + key.ciphertext, + key.iv, + key.tag, + "user_installer", + ], + ) +} + +const makeConfig = (tokens: { slack?: string; shared?: string } = {}) => + ConfigProvider.layer( + ConfigProvider.fromUnknown({ + PORT: "3472", + TINYBIRD_HOST: "https://api.tinybird.co", + TINYBIRD_TOKEN: "test-token", + MAPLE_AUTH_MODE: "self_hosted", + MAPLE_ROOT_PASSWORD: "test-root-password", + MAPLE_DEFAULT_ORG_ID: "default", + MAPLE_INGEST_KEY_ENCRYPTION_KEY: ENCRYPTION_KEY.toString("base64"), + MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: "maple-test-lookup-secret", + MAPLE_APP_BASE_URL: "https://web.localhost", + ...(tokens.slack !== undefined ? { SLACK_INTERNAL_SERVICE_TOKEN: tokens.slack } : {}), + ...(tokens.shared !== undefined ? { INTERNAL_SERVICE_TOKEN: tokens.shared } : {}), + }), + ) + +const makeRouterLayer = (testDb: TestDb, tokens: { slack?: string; shared?: string } = {}) => + SlackInternalRouter.pipe( + Layer.provide(SlackIntegrationService.layer), + Layer.provide(Layer.mergeAll(ApiKeysService.layer, OAuthStateRepository.layer)), + Layer.provide(testDb.layer), + Layer.provide(Env.layer), + Layer.provide(makeConfig(tokens)), + ) + +const TEAM_PATH = "/internal/slack/workspaces" + +const get = ( + handler: (request: Request) => Promise, + teamId: string, + bearer?: string, +) => + Effect.promise(() => + handler( + new Request(`http://api.localhost${TEAM_PATH}/${teamId}`, { + headers: bearer !== undefined ? { authorization: bearer } : {}, + }), + ), + ) + +/** Run `body` against a web handler for the internal router, disposing after. */ +const withHandler = ( + testDb: TestDb, + tokens: { slack?: string; shared?: string }, + body: (handler: (request: Request) => Promise) => Effect.Effect, +) => { + const { handler, dispose } = HttpRouter.toWebHandler(makeRouterLayer(testDb, tokens), { + disableLogger: true, + }) + return body((request) => handler(request)).pipe(Effect.ensuring(Effect.promise(dispose))) +} + +describe("SlackInternalRouter", () => { + it.effect("resolves a team with the dedicated token and returns the fixed contract", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_1", + orgId: "org_a", + teamId: "T0123", + teamName: "Acme", + botToken: "xoxb-secret-token", + apiKey: "maple_ak_secret", + }), + ) + yield* withHandler( + testDb, + { slack: "slack-secret-token", shared: "shared-secret-token" }, + Effect.fnUntraced(function* (handler) { + const ok = yield* get(handler, "T0123", "Bearer maple_svc_slack-secret-token") + assert.strictEqual(ok.status, 200) + const body = yield* Effect.promise(() => ok.json()) + // FIXED response contract — the Railway bot is built against exactly + // these keys (SlackBotResolutionResponseSchema). + assert.deepStrictEqual(body, { + orgId: "org_a", + teamId: "T0123", + teamName: "Acme", + botToken: "xoxb-secret-token", + mapleApiKey: "maple_ak_secret", + }) + + // When the dedicated token is set, the shared token is NOT accepted. + const shared = yield* get(handler, "T0123", "Bearer maple_svc_shared-secret-token") + assert.strictEqual(shared.status, 401) + }), + ) + // Building testDb.layer (provided below) applies the migrations before + // the raw-SQL insert above runs. + }).pipe(Effect.provide(testDb.layer)) + }) + + it.effect("falls back to INTERNAL_SERVICE_TOKEN when the Slack-specific token is unset", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_2", + orgId: "org_b", + teamId: "T0999", + teamName: "Beta", + botToken: "xoxb-b", + apiKey: "maple_ak_b", + }), + ) + yield* withHandler( + testDb, + { shared: "shared-only-token" }, + Effect.fnUntraced(function* (handler) { + const ok = yield* get(handler, "T0999", "Bearer maple_svc_shared-only-token") + assert.strictEqual(ok.status, 200) + }), + ) + }).pipe(Effect.provide(testDb.layer)) + }) + + it.effect("rejects bad or missing credentials with 401", () => { + const testDb = createTestDb(trackedDbs) + return withHandler( + testDb, + { slack: "slack-secret-token" }, + Effect.fnUntraced(function* (handler) { + // Wrong token of the SAME length (timingSafeEqual path). + const wrong = yield* get(handler, "T0123", `Bearer maple_svc_${"x".repeat("slack-secret-token".length)}`) + assert.strictEqual(wrong.status, 401) + + // Length mismatch must 401 cleanly, not throw out of timingSafeEqual. + const short = yield* get(handler, "T0123", "Bearer maple_svc_nope") + assert.strictEqual(short.status, 401) + + // Missing the maple_svc_ prefix. + const unprefixed = yield* get(handler, "T0123", "Bearer slack-secret-token") + assert.strictEqual(unprefixed.status, 401) + + // Wrong scheme / missing header. + const basic = yield* get(handler, "T0123", "Basic maple_svc_slack-secret-token") + assert.strictEqual(basic.status, 401) + const missing = yield* get(handler, "T0123") + assert.strictEqual(missing.status, 401) + }), + ) + }) + + it.effect("rejects every request with 401 when no internal token is configured", () => { + const testDb = createTestDb(trackedDbs) + return withHandler( + testDb, + {}, + Effect.fnUntraced(function* (handler) { + const response = yield* get(handler, "T0123", "Bearer maple_svc_anything") + assert.strictEqual(response.status, 401) + const text = yield* Effect.promise(() => response.text()) + assert.include(text, "not configured") + }), + ) + }) + + it.effect("returns 404 for unknown or revoked teams and 400 for an invalid teamId", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_3", + orgId: "org_c", + teamId: "T-revoked", + teamName: "Gone", + botToken: "xoxb-c", + apiKey: "maple_ak_c", + revoked: true, + }), + ) + yield* withHandler( + testDb, + { slack: "slack-secret-token" }, + Effect.fnUntraced(function* (handler) { + const bearer = "Bearer maple_svc_slack-secret-token" + + const unknown = yield* get(handler, "T-unknown", bearer) + assert.strictEqual(unknown.status, 404) + + const revoked = yield* get(handler, "T-revoked", bearer) + assert.strictEqual(revoked.status, 404) + + // "%20" decodes to a lone space — fails the trimmed/non-empty check. + const invalid = yield* get(handler, "%20", bearer) + assert.strictEqual(invalid.status, 400) + }), + ) + }).pipe(Effect.provide(testDb.layer)) + }) +}) diff --git a/apps/api/src/routes/slack-integration.http.ts b/apps/api/src/routes/slack-integration.http.ts new file mode 100644 index 00000000..d32bdb86 --- /dev/null +++ b/apps/api/src/routes/slack-integration.http.ts @@ -0,0 +1,187 @@ +import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { Effect, Option, Redacted, Schema } from "effect" +import { timingSafeEqual } from "node:crypto" +import { Env } from "../lib/Env" +import { + SlackIntegrationService, + SlackBotResolutionResponseSchema, + SLACK_CALLBACK_PATH, +} from "../services/SlackIntegrationService" + +const INTERNAL_SERVICE_PREFIX = "maple_svc_" + +/** Redirect target on the web app after an install attempt. */ +const buildAppRedirect = (appBaseUrl: string, params: Record): string => { + const base = appBaseUrl.replace(/\/$/, "") + // Land on the Slack integration card (route `/integrations`, search `integration`), + // carrying the `slack=connected|error` return params the card surfaces as a toast. + const search = new URLSearchParams({ integration: "slack", ...params }).toString() + return `${base}/integrations?${search}` +} + +/** + * Public Slack OAuth callback (`GET /oauth/slack/callback`). Slack redirects the + * browser here after the user approves (or denies) the install; we exchange the + * code, persist the workspace, then redirect the browser back to the web app's + * integrations page with a success/error query param. + */ +export const SlackCallbackRouter = HttpRouter.use((router) => + Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const env = yield* Env + + const redirect = (params: Record) => + HttpServerResponse.redirect(buildAppRedirect(env.MAPLE_APP_BASE_URL, params)) + + const handle = Effect.fn("slack.oauthCallback")(function* ( + req: HttpServerRequest.HttpServerRequest, + ) { + const urlOption = Option.liftThrowable(() => new URL(req.url, "http://localhost"))() + if (Option.isNone(urlOption)) { + return redirect({ slack: "error", slack_message: "Malformed callback URL" }) + } + const url = urlOption.value + const code = url.searchParams.get("code") + const state = url.searchParams.get("state") + const oauthError = url.searchParams.get("error") + + if (oauthError) { + return redirect({ slack: "error", slack_message: oauthError }) + } + if (!code || !state) { + return redirect({ slack: "error", slack_message: "Missing code or state in callback" }) + } + + return yield* slack.completeInstall(code, state).pipe( + Effect.tapError((error) => + Effect.logError("Slack OAuth completeInstall failed", { + tag: error._tag, + message: error.message, + }), + ), + Effect.map((result) => + redirect({ + slack: "connected", + ...(result.teamName ? { slack_team: result.teamName } : {}), + }), + ), + Effect.catchTags({ + "@maple/http/errors/IntegrationsValidationError": (error) => + Effect.succeed(redirect({ slack: "error", slack_message: error.message })), + "@maple/http/errors/IntegrationsForbiddenError": (error) => + Effect.succeed(redirect({ slack: "error", slack_message: error.message })), + "@maple/http/errors/IntegrationsUpstreamError": () => + Effect.succeed( + redirect({ slack: "error", slack_message: "Failed to complete the Slack connection" }), + ), + "@maple/http/errors/IntegrationsPersistenceError": () => + Effect.succeed( + redirect({ slack: "error", slack_message: "Failed to complete the Slack connection" }), + ), + }), + ) + }) + + yield* router.add("GET", SLACK_CALLBACK_PATH, handle) + }), +) + +const errorText = (message: string, status: number) => + HttpServerResponse.text(message, { status, headers: { "content-type": "text/plain; charset=utf-8" } }) + +/** + * Constant-time check of an `Authorization: Bearer maple_svc_` header + * against the configured `INTERNAL_SERVICE_TOKEN`. Mirrors + * `resolveMcpTenantContext`'s internal-service auth. + */ +const isValidServiceBearer = ( + authorization: string | undefined, + internalToken: string | undefined, +): boolean => { + if (!internalToken) return false + if (!authorization) return false + const [scheme, token] = authorization.split(" ") + if (!scheme || !token || scheme.toLowerCase() !== "bearer") return false + if (!token.startsWith(INTERNAL_SERVICE_PREFIX)) return false + const provided = token.slice(INTERNAL_SERVICE_PREFIX.length) + return ( + provided.length === internalToken.length && + timingSafeEqual(Buffer.from(provided), Buffer.from(internalToken)) + ) +} + +/** Non-empty, trimmed `:teamId` path param. */ +const decodeTeamIdParam = Schema.decodeUnknownOption( + Schema.String.pipe(Schema.check(Schema.isMinLength(1), Schema.isTrimmed())), +) + +const encodeBotResolution = Schema.encodeUnknownEffect(SlackBotResolutionResponseSchema) + +/** + * Internal endpoint for the Railway-hosted Slack bot. Given a Slack `teamId`, + * returns the bound org's decrypted bot token + minted Maple API key so the bot + * can act on the org's behalf. Guarded by a dedicated internal-service token + * (`Authorization: Bearer maple_svc_`), falling + * back to the shared `INTERNAL_SERVICE_TOKEN` when the dedicated one is unset. + * + * Response contract (FIXED — the bot is built against it): + * 200 → { orgId, teamId, teamName, botToken, mapleApiKey } + * 404 → unknown or revoked team + */ +export const SlackInternalRouter = HttpRouter.use((router) => + Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const env = yield* Env + // Prefer the Slack-bot-specific secret so the Railway bot can hold a token + // distinct from the MCP-internal one; fall back to the shared token. + const internalToken = Option.match( + Option.orElse(env.SLACK_INTERNAL_SERVICE_TOKEN, () => env.INTERNAL_SERVICE_TOKEN), + { + onNone: () => undefined, + onSome: Redacted.value, + }, + ) + + const logAccess = (teamId: string, outcome: "found" | "not-found" | "unauthorized") => + Effect.logInfo("Slack internal resolve access", { teamId, outcome }) + + const handle = Effect.fn("slack.internalResolve")(function* ( + req: HttpServerRequest.HttpServerRequest, + ) { + const params = yield* HttpRouter.params + const teamIdOption = decodeTeamIdParam( + typeof params.teamId === "string" ? decodeURIComponent(params.teamId) : undefined, + ) + const teamId = Option.getOrElse(teamIdOption, () => "") + if (!internalToken) { + yield* logAccess(teamId, "unauthorized") + return errorText("Internal service token is not configured", 401) + } + if (!isValidServiceBearer(req.headers.authorization, internalToken)) { + yield* logAccess(teamId, "unauthorized") + return errorText("Unauthorized", 401) + } + if (Option.isNone(teamIdOption)) return errorText("Missing teamId", 400) + + const resolution = yield* slack.resolveForBot(teamIdOption.value).pipe( + Effect.map(Option.some), + Effect.catchTag("@maple/http/errors/IntegrationsNotConnectedError", () => + Effect.succeedNone, + ), + ) + return yield* Option.match(resolution, { + onNone: () => + logAccess(teamId, "not-found").pipe( + Effect.as(errorText("No active Slack installation for this team", 404)), + ), + onSome: (resolved) => + logAccess(teamId, "found").pipe( + Effect.andThen(encodeBotResolution(resolved).pipe(Effect.orDie)), + Effect.flatMap((encoded) => HttpServerResponse.json(encoded)), + ), + }) + }) + + yield* router.add("GET", "/internal/slack/workspaces/:teamId", handle) + }), +) diff --git a/apps/api/src/routes/v2/alert-destinations.http.ts b/apps/api/src/routes/v2/alert-destinations.http.ts index d61d38d7..81ac0e6c 100644 --- a/apps/api/src/routes/v2/alert-destinations.http.ts +++ b/apps/api/src/routes/v2/alert-destinations.http.ts @@ -8,6 +8,7 @@ import { HazelOAuthAlertDestinationConfig, PagerDutyAlertDestinationConfig, SlackAlertDestinationConfig, + SlackBotAlertDestinationConfig, WebhookAlertDestinationConfig, } from "@maple/domain/http" import type { @@ -51,6 +52,14 @@ const toCreateRequest = (params: V2AlertDestinationCreateParams) => { ...(params.channel_label !== undefined ? { channelLabel: params.channel_label } : {}), ...(params.enabled !== undefined ? { enabled: params.enabled } : {}), }) + case "slack-bot": + return new SlackBotAlertDestinationConfig({ + type: "slack-bot", + name: params.name, + channelId: params.channel_id, + ...(params.channel_name !== undefined ? { channelName: params.channel_name } : {}), + ...(params.enabled !== undefined ? { enabled: params.enabled } : {}), + }) case "pagerduty": return new PagerDutyAlertDestinationConfig({ type: "pagerduty", @@ -117,6 +126,13 @@ const toUpdateRequest = (params: V2AlertDestinationUpdateParams): AlertDestinati ...(params.webhook_url !== undefined ? { webhookUrl: params.webhook_url } : {}), ...(params.channel_label !== undefined ? { channelLabel: params.channel_label } : {}), } + case "slack-bot": + return { + type: "slack-bot", + ...shared, + ...(params.channel_id !== undefined ? { channelId: params.channel_id } : {}), + ...(params.channel_name !== undefined ? { channelName: params.channel_name } : {}), + } case "pagerduty": return { type: "pagerduty", diff --git a/apps/api/src/routes/v2/integrations-origin.test.ts b/apps/api/src/routes/v2/integrations-origin.test.ts new file mode 100644 index 00000000..18d8ad29 --- /dev/null +++ b/apps/api/src/routes/v2/integrations-origin.test.ts @@ -0,0 +1,62 @@ +import type { HttpServerRequest } from "effect/unstable/http" +import { assert, describe, it } from "@effect/vitest" +import { resolveRequestOrigin } from "./integrations.http" + +/** + * `resolveRequestOrigin` is a pure header/url function, so it is tested with a + * minimal structural fake — it only reads `headers` and `url`. + */ +const fakeRequest = ( + headers: Record, + url = "/v2/integrations/slack/install", +): HttpServerRequest.HttpServerRequest => + ({ headers, url }) as unknown as HttpServerRequest.HttpServerRequest + +describe("resolveRequestOrigin", () => { + it("uses x-forwarded-host and x-forwarded-proto when both are present", () => { + const origin = resolveRequestOrigin( + fakeRequest({ "x-forwarded-host": "api.maple.dev", "x-forwarded-proto": "https", host: "internal:8787" }), + ) + assert.strictEqual(origin, "https://api.maple.dev") + }) + + it("prefers the forwarded host over the direct host header", () => { + const origin = resolveRequestOrigin( + fakeRequest({ "x-forwarded-host": "public.example.com", host: "10.0.0.5:8080" }), + ) + assert.strictEqual(origin, "https://public.example.com") + }) + + it("respects a forwarded proto of http even for a non-localhost host", () => { + const origin = resolveRequestOrigin( + fakeRequest({ "x-forwarded-host": "api.maple.dev", "x-forwarded-proto": "http" }), + ) + assert.strictEqual(origin, "http://api.maple.dev") + }) + + it("falls back to the host header and defaults to https for non-local hosts", () => { + const origin = resolveRequestOrigin(fakeRequest({ host: "api.maple.dev" })) + assert.strictEqual(origin, "https://api.maple.dev") + }) + + it("defaults to http for localhost and 127.* hosts, preserving the port", () => { + assert.strictEqual(resolveRequestOrigin(fakeRequest({ host: "localhost:3472" })), "http://localhost:3472") + assert.strictEqual(resolveRequestOrigin(fakeRequest({ host: "127.0.0.1:3472" })), "http://127.0.0.1:3472") + assert.strictEqual(resolveRequestOrigin(fakeRequest({ host: "localhost" })), "http://localhost") + }) + + it("preserves an explicit port on a non-local host", () => { + const origin = resolveRequestOrigin(fakeRequest({ host: "api.staging.maple.dev:8443" })) + assert.strictEqual(origin, "https://api.staging.maple.dev:8443") + }) + + it("falls back to parsing an absolute request url when no host headers exist", () => { + const origin = resolveRequestOrigin(fakeRequest({}, "https://api.example.com:8443/v2/integrations/slack")) + assert.strictEqual(origin, "https://api.example.com:8443") + }) + + it("returns an empty origin when no host is known and the url is relative", () => { + const origin = resolveRequestOrigin(fakeRequest({}, "/v2/integrations/slack/install")) + assert.strictEqual(origin, "") + }) +}) diff --git a/apps/api/src/routes/v2/integrations.http.ts b/apps/api/src/routes/v2/integrations.http.ts new file mode 100644 index 00000000..27b4c515 --- /dev/null +++ b/apps/api/src/routes/v2/integrations.http.ts @@ -0,0 +1,149 @@ +import { HttpServerRequest } from "effect/unstable/http" +import { HttpApiBuilder } from "effect/unstable/httpapi" +import { CurrentTenant } from "@maple/domain/http" +import type { + V2SlackChannelList, + V2SlackIntegrationStatus, + V2SlackInstallResponse, + V2SlackUninstallResponse, +} from "@maple/domain/http/v2" +import { + MapleApiV2, + isoTimestampOrNull, + notFound, + permissionError, + serviceUnavailable, + upstreamError, +} from "@maple/domain/http/v2" +import { Effect, Option } from "effect" +import { requireAdmin } from "../../lib/auth" +import type { SlackChannelSummary, SlackInstallStatus } from "../../services/SlackIntegrationService" +import { SLACK_CALLBACK_PATH, SlackIntegrationService } from "../../services/SlackIntegrationService" + +export const resolveRequestOrigin = (req: HttpServerRequest.HttpServerRequest): string => { + const headers = req.headers as Record + const forwardedHost = headers["x-forwarded-host"] + const forwardedProto = headers["x-forwarded-proto"] + const host = forwardedHost ?? headers.host + if (host) { + const proto = + forwardedProto ?? (host.startsWith("localhost") || host.startsWith("127.") ? "http" : "https") + return `${proto}://${host}` + } + return Option.match(Option.liftThrowable(() => new URL(req.url))(), { + onNone: () => "", + onSome: (parsed) => `${parsed.protocol}//${parsed.host}`, + }) +} + +const toStatus = (status: SlackInstallStatus): V2SlackIntegrationStatus => ({ + object: "slack_integration", + installed: status.installed, + team_id: status.teamId, + team_name: status.teamName, + bot_user_id: status.botUserId, + installed_at: isoTimestampOrNull(status.installedAt), +}) + +const toChannelList = (channels: ReadonlyArray): V2SlackChannelList => ({ + object: "slack_integration.channel_list", + channels: channels.map((channel) => ({ + id: channel.id, + name: channel.name, + is_private: channel.isPrivate, + is_member: channel.isMember, + })), +}) + +export const HttpV2SlackIntegrationsLive = HttpApiBuilder.group(MapleApiV2, "slackIntegration", (handlers) => + Effect.gen(function* () { + const slack = yield* SlackIntegrationService + + return handlers + .handle("status", () => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + const status = yield* slack.getStatus(tenant.orgId).pipe( + Effect.tapError((error) => + Effect.logError("Slack integration status failed", { + tag: error._tag, + message: error.message, + }), + ), + Effect.catchTags({ + "@maple/http/errors/IntegrationsPersistenceError": (error) => + Effect.fail(serviceUnavailable(error.message)), + }), + ) + return toStatus(status) + }), + ) + .handle("install", () => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* requireAdmin(tenant.roles, () => + permissionError( + "insufficient_permissions", + "Only org admins can install the Slack app", + ), + ) + const req = yield* HttpServerRequest.HttpServerRequest + const callbackUrl = `${resolveRequestOrigin(req)}${SLACK_CALLBACK_PATH}` + const result = yield* slack + .startInstall(tenant.orgId, tenant.userId, callbackUrl) + .pipe( + Effect.catchTags({ + "@maple/http/errors/IntegrationsValidationError": (error) => + Effect.fail(serviceUnavailable(error.message)), + "@maple/http/errors/IntegrationsPersistenceError": (error) => + Effect.fail(serviceUnavailable(error.message)), + }), + ) + return { object: "slack_integration.install" as const, url: result.url } satisfies V2SlackInstallResponse + }), + ) + .handle("uninstall", () => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* requireAdmin(tenant.roles, () => + permissionError( + "insufficient_permissions", + "Only org admins can uninstall the Slack app", + ), + ) + yield* slack.uninstall(tenant.orgId).pipe( + Effect.tapError((error) => + Effect.logError("Slack integration uninstall failed", { + tag: error._tag, + message: error.message, + }), + ), + Effect.catchTags({ + "@maple/http/errors/IntegrationsPersistenceError": (error) => + Effect.fail(serviceUnavailable(error.message)), + }), + ) + return { + object: "slack_integration" as const, + installed: false as const, + } satisfies V2SlackUninstallResponse + }), + ) + .handle("channels", () => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + const channels = yield* slack.listChannels(tenant.orgId).pipe( + Effect.catchTags({ + "@maple/http/errors/IntegrationsNotConnectedError": (error) => + Effect.fail(notFound(error.message)), + "@maple/http/errors/IntegrationsUpstreamError": (error) => + Effect.fail(upstreamError("slack_upstream_error", error.message)), + "@maple/http/errors/IntegrationsPersistenceError": (error) => + Effect.fail(serviceUnavailable(error.message)), + }), + ) + return toChannelList(channels) + }), + ) + }), +) diff --git a/apps/api/src/routes/v2/v2-test-support.ts b/apps/api/src/routes/v2/v2-test-support.ts index 0c8dcad9..45f29731 100644 --- a/apps/api/src/routes/v2/v2-test-support.ts +++ b/apps/api/src/routes/v2/v2-test-support.ts @@ -8,6 +8,7 @@ import { OrganizationService } from "../../services/OrganizationService" import { OrgIngestKeysService } from "../../services/OrgIngestKeysService" import { RecommendationIssueService } from "../../services/RecommendationIssueService" import { ScrapeTargetsService } from "../../services/ScrapeTargetsService" +import { SlackIntegrationService } from "../../services/SlackIntegrationService" import { ApiV2RateLimiter } from "../../services/ApiV2RateLimiter" import { WarehouseQueryService } from "../../lib/WarehouseQueryService" import { QueryEngineService } from "../../services/QueryEngineService" @@ -18,6 +19,7 @@ import { HttpV2ApiKeysLive } from "./api-keys.http" import { HttpV2AttributeMappingsLive } from "./attribute-mappings.http" import { HttpV2DashboardsLive } from "./dashboards.http" import { HttpV2IngestKeysLive } from "./ingest-keys.http" +import { HttpV2SlackIntegrationsLive } from "./integrations.http" import { HttpV2ErrorIssuesLive } from "./error-issues.http" import { HttpV2AnomaliesLive } from "./anomalies.http" import { HttpV2InvestigationsLive } from "./investigations.http" @@ -40,8 +42,33 @@ import { * the groups it does not exercise. */ +/** + * Inert SlackIntegrationService — the slack integration group is registered on + * MapleApiV2, so the full-API harnesses must satisfy it even when they don't + * exercise it. Provided directly onto the group layer below so it leaks no + * requirement into the harnesses. + */ +const SlackIntegrationServiceStubLayer = Layer.succeed( + SlackIntegrationService, + SlackIntegrationService.of({ + startInstall: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + completeInstall: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + getStatus: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + uninstall: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + listChannels: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + resolveForBot: () => + Effect.die(new Error("SlackIntegrationService is not available in this test harness")), + }), +) + export const AllV2GroupLayersLive = Layer.mergeAll( HttpV2ApiKeysLive, + HttpV2SlackIntegrationsLive.pipe(Layer.provide(SlackIntegrationServiceStubLayer)), HttpV2DashboardsLive, HttpV2AlertRulesLive, HttpV2AlertDestinationsLive, diff --git a/apps/api/src/services/AlertDeliveryDispatch.test.ts b/apps/api/src/services/AlertDeliveryDispatch.test.ts index 69bfed12..c140b59f 100644 --- a/apps/api/src/services/AlertDeliveryDispatch.test.ts +++ b/apps/api/src/services/AlertDeliveryDispatch.test.ts @@ -1,7 +1,8 @@ import type { AlertDestinationRow } from "@maple/db" import { AlertDeliveryError, AlertDestinationId } from "@maple/domain/http" import { assert, describe, it } from "@effect/vitest" -import { Effect, Schema } from "effect" +import { Effect, Fiber, Schema } from "effect" +import { TestClock } from "effect/testing" import { buildAlertChatUrl, buildDiscordEmbedsFromTemplate, @@ -39,10 +40,17 @@ const LINK = "https://web.localhost/alerts" const CHAT = "https://web.localhost/chat?mode=alert" const DESTINATION_ID = Schema.decodeUnknownSync(AlertDestinationId)("7c6b5a49-3821-4e0f-9d8c-7b6a59483726") +/** Slack-token resolver that must not be invoked for non-slack-bot destinations. */ +const failingSlackToken = () => + Effect.fail( + new AlertDeliveryError({ message: "unexpected resolveSlackBotToken", destinationType: "slack-bot" }), + ) + /** Dispatch deps for non-email destinations — email sends must not happen. */ const noEmailDeps: DispatchDeps = { sendEmail: () => Effect.fail(new AlertDeliveryError({ message: "unexpected sendEmail", destinationType: "email" })), + resolveSlackBotToken: failingSlackToken, } describe("buildAlertChatUrl (Ask Maple AI link)", () => { @@ -198,6 +206,160 @@ describe("dispatchDelivery", () => { }), ) + const slackBotContext: DispatchContext = { + ...pagerdutyContext, + destination: { ...destinationRow, name: "Slack bot", type: "slack-bot" }, + publicConfig: { summary: "#incidents", channelLabel: "#incidents" }, + secretConfig: { type: "slack-bot", channelId: "C0789CHAN", channelName: "incidents" }, + } + + const slackTokenDeps = (token = "xoxb-test-token"): DispatchDeps => ({ + sendEmail: () => + Effect.fail(new AlertDeliveryError({ message: "unexpected sendEmail", destinationType: "email" })), + resolveSlackBotToken: () => Effect.succeed(token), + }) + + it.effect("slack-bot: posts to chat.postMessage with the resolved bot token + channel", () => + Effect.gen(function* () { + const calls: Array<{ url: string; auth: string | null; body: unknown }> = [] + const fetchFn: typeof fetch = async (input, init) => { + calls.push({ + url: String(input), + auth: new Headers(init?.headers).get("authorization"), + body: JSON.parse(String(init?.body)), + }) + return new Response(JSON.stringify({ ok: true, ts: "1700000000.000100" }), { status: 200 }) + } + + const result = yield* dispatchDelivery( + slackBotContext, + "{}", + fetchFn, + 5_000, + LINK, + CHAT, + slackTokenDeps(), + ) + + assert.strictEqual(calls.length, 1) + assert.strictEqual(calls[0]!.url, "https://slack.com/api/chat.postMessage") + assert.strictEqual(calls[0]!.auth, "Bearer xoxb-test-token") + assert.strictEqual((calls[0]!.body as { channel: string }).channel, "C0789CHAN") + assert.isArray((calls[0]!.body as { blocks: unknown[] }).blocks) + assert.strictEqual(result.providerReference, "1700000000.000100") + assert.strictEqual(result.responseCode, 200) + }), + ) + + it.effect("slack-bot: surfaces a not_in_channel logical error with an actionable message", () => + Effect.gen(function* () { + const fetchFn: typeof fetch = async () => + new Response(JSON.stringify({ ok: false, error: "not_in_channel" }), { status: 200 }) + + const error = yield* Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, slackTokenDeps()), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "not_in_channel") + assert.include(error.message, "invite the Maple bot") + }), + ) + + it.effect("slack-bot: a hung fetch times out via the Clock and fails typed", () => + Effect.gen(function* () { + // Never settles — only the Clock-driven timeoutOrElse can end this. + const fetchFn: typeof fetch = () => new Promise(() => {}) + + const fiber = yield* Effect.forkChild( + Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, slackTokenDeps()), + ), + { startImmediately: true }, + ) + yield* TestClock.adjust("6 seconds") + const error = yield* Fiber.join(fiber) + + assert.instanceOf(error, AlertDeliveryError) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "timed out after 5000ms") + }), + ) + + it.effect("slack-bot: a non-JSON 200 response fails typed", () => + Effect.gen(function* () { + const fetchFn: typeof fetch = async () => new Response("gateway says hi", { status: 200 }) + + const error = yield* Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, slackTokenDeps()), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "non-JSON response") + }), + ) + + it.effect("slack-bot: a JSON payload that fails the response schema fails typed", () => + Effect.gen(function* () { + // Valid JSON, but `ok` is a string — SlackPostMessageResponseSchema rejects it. + const fetchFn: typeof fetch = async () => + new Response(JSON.stringify({ ok: "yes", ts: 123 }), { status: 200 }) + + const error = yield* Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, slackTokenDeps()), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "unexpected response payload") + }), + ) + + it.effect("slack-bot: an HTTP-level failure surfaces the status and body", () => + Effect.gen(function* () { + const fetchFn: typeof fetch = async () => new Response("upstream exploded", { status: 500 }) + + const error = yield* Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, slackTokenDeps()), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "Slack delivery failed with 500") + assert.include(error.message, "upstream exploded") + }), + ) + + it.effect("slack-bot: fails when the org has no active Slack installation", () => + Effect.gen(function* () { + const fetchFn: typeof fetch = async () => { + throw new Error("fetch must not be called when token resolution fails") + } + const deps: DispatchDeps = { + sendEmail: () => + Effect.fail( + new AlertDeliveryError({ message: "unexpected sendEmail", destinationType: "email" }), + ), + resolveSlackBotToken: () => + Effect.fail( + new AlertDeliveryError({ + message: "Slack is not connected for this organization", + destinationType: "slack-bot", + }), + ), + } + + const error = yield* Effect.flip( + dispatchDelivery(slackBotContext, "{}", fetchFn, 5_000, LINK, CHAT, deps), + ) + + assert.instanceOf(error, AlertDeliveryError) + assert.include(error.message, "Slack is not connected") + }), + ) + const failingFetch: typeof fetch = async () => { throw new Error("fetch must not be called for email dispatch") } @@ -222,6 +384,13 @@ describe("dispatchDelivery", () => { Effect.sync(() => { sent.push({ to, subject, html }) }), + resolveSlackBotToken: () => + Effect.fail( + new AlertDeliveryError({ + message: "resolveSlackBotToken not available in this test", + destinationType: "slack-bot", + }), + ), } const result = yield* dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps) @@ -250,6 +419,7 @@ describe("dispatchDelivery", () => { destinationType: "email", }), ), + resolveSlackBotToken: failingSlackToken, } const error = yield* Effect.flip( @@ -277,6 +447,7 @@ describe("dispatchDelivery", () => { : Effect.sync(() => { sent.push(to) }), + resolveSlackBotToken: failingSlackToken, } const result = yield* dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps) @@ -298,6 +469,7 @@ describe("dispatchDelivery", () => { destinationType: "email", }), ), + resolveSlackBotToken: failingSlackToken, } const error = yield* Effect.flip( diff --git a/apps/api/src/services/AlertDeliveryDispatch.ts b/apps/api/src/services/AlertDeliveryDispatch.ts index 25d82127..b2bbdad0 100644 --- a/apps/api/src/services/AlertDeliveryDispatch.ts +++ b/apps/api/src/services/AlertDeliveryDispatch.ts @@ -8,7 +8,7 @@ import { type AlertSignalType, } from "@maple/domain/http" import type { AlertDestinationRow } from "@maple/db" -import { Clock, Duration, Effect, Match, Option } from "effect" +import { Clock, Duration, Effect, Match, Option, Schema } from "effect" import type { EnrichedDestinationSecretConfig } from "./AlertDestinationHydration" import { safeFetch } from "../lib/url-validator" // Circular with ./alert-email (it imports our formatting helpers); safe — both @@ -247,6 +247,18 @@ export const formatObservedSummary = (context: ObservedContext): string => { const makeDeliveryError = (message: string, destinationType?: AlertDestinationType) => new AlertDeliveryError({ message, destinationType }) +/** + * Slack Web API `chat.postMessage` response envelope. Slack returns HTTP 200 + * with `{ ok: false, error }` on logical failures, so the body is the source + * of truth. + */ +const SlackPostMessageResponseSchema = Schema.Struct({ + ok: Schema.optionalKey(Schema.Boolean), + error: Schema.optionalKey(Schema.String), + ts: Schema.optionalKey(Schema.String), +}) +const decodeSlackPostMessageResponse = Schema.decodeUnknownEffect(SlackPostMessageResponseSchema) + const runTimedFetch = ( destinationType: AlertDestinationType, label: string, @@ -572,6 +584,13 @@ export interface DispatchDeps { * binding). Injected like `fetchFn` so this module stays dependency-free. */ readonly sendEmail: (to: string, subject: string, html: string) => Effect.Effect + /** + * Resolves the decrypted Slack bot token for an org from its + * `slack_workspaces` row. Injected like `sendEmail` so this module stays + * dependency-free — only the `slack-bot` destination arm invokes it. Fails + * (AlertDeliveryError) when the org has no active Slack installation. + */ + readonly resolveSlackBotToken: (orgId: string) => Effect.Effect } export const dispatchDelivery = ( @@ -631,6 +650,79 @@ export const dispatchDelivery = ( responseCode: response.status, } as DispatchResult }), + "slack-bot": (config) => + Effect.gen(function* () { + const botToken = yield* deps.resolveSlackBotToken(context.destination.orgId) + const templated = renderTitleBody(context, "slack", linkUrl, chatUrl) + const blocks = templated + ? buildSlackBlocksFromTemplate( + templated.title, + templated.body, + context, + linkUrl, + chatUrl, + ) + : buildSlackBlocks(context, linkUrl, chatUrl) + const response = yield* runTimedFetch( + "slack-bot", + "Slack", + fetchFn, + timeoutMs, + () => + fetchFn("https://slack.com/api/chat.postMessage", { + method: "POST", + headers: { + "content-type": "application/json; charset=utf-8", + authorization: `Bearer ${botToken}`, + }, + body: JSON.stringify({ + channel: config.channelId, + text: + templated?.title ?? + `${context.ruleName}: ${formatEventTypeLabel(context.eventType)}`, + blocks, + }), + }), + ) + if (!response.ok) { + const detail = yield* readErrorBody(response) + return yield* Effect.fail( + makeDeliveryError( + `Slack delivery failed with ${response.status}${detail ? `: ${detail}` : ""}`, + "slack-bot", + ), + ) + } + // Slack Web API returns HTTP 200 with `{ ok: false, error }` on + // logical failures, so the JSON body — not the status — is the + // source of truth. + const rawPayload = yield* Effect.tryPromise({ + try: () => response.json(), + catch: () => + makeDeliveryError("Slack returned a non-JSON response", "slack-bot"), + }) + const payload = yield* decodeSlackPostMessageResponse(rawPayload).pipe( + Effect.mapError((error) => + makeDeliveryError( + `Slack returned an unexpected response payload: ${error.message}`, + "slack-bot", + ), + ), + ) + if (!payload.ok) { + const error = payload.error ?? "unknown_error" + const message = + error === "not_in_channel" || error === "channel_not_found" + ? `Slack rejected the message (${error}) — invite the Maple bot to the channel and try again` + : `Slack rejected the message: ${error}` + return yield* Effect.fail(makeDeliveryError(message, "slack-bot")) + } + return { + providerMessage: `Delivered to Slack #${config.channelName ?? config.channelId}`, + providerReference: payload.ts ?? null, + responseCode: response.status, + } as DispatchResult + }), pagerduty: (config) => Effect.gen(function* () { const templated = renderTitleBody(context, "pagerduty", linkUrl, chatUrl) diff --git a/apps/api/src/services/AlertDestinationHydration.ts b/apps/api/src/services/AlertDestinationHydration.ts index 6679c9e5..1b366486 100644 --- a/apps/api/src/services/AlertDestinationHydration.ts +++ b/apps/api/src/services/AlertDestinationHydration.ts @@ -18,6 +18,13 @@ const DestinationSecretConfigSchema = Schema.Union([ type: Schema.Literal("slack"), webhookUrl: Schema.String, }), + Schema.Struct({ + type: Schema.Literal("slack-bot"), + // No secret token here — the bot token is resolved from the org's + // slack_workspaces row at dispatch time. Only the target channel is stored. + channelId: Schema.String, + channelName: Schema.NullOr(Schema.String), + }), Schema.Struct({ type: Schema.Literal("pagerduty"), integrationKey: Schema.String, diff --git a/apps/api/src/services/AlertsService.test.ts b/apps/api/src/services/AlertsService.test.ts index fb6f8449..7947cdba 100644 --- a/apps/api/src/services/AlertsService.test.ts +++ b/apps/api/src/services/AlertsService.test.ts @@ -30,6 +30,7 @@ import { EmailService } from "../lib/EmailService" import { OrgMembersError, OrgMembersService, type OrgMember } from "./OrgMembersService" import { QueryEngineService } from "./QueryEngineService" import { cleanupTestDbs, createTestDb, executeSql, queryFirstRow, type TestDb } from "../lib/test-pglite" +import { decryptAes256Gcm } from "../lib/Crypto" const trackedDbs: TestDb[] = [] @@ -1824,6 +1825,98 @@ describe("AlertsService", () => { ) }) + it.effect("slack-bot update merges channel fields into the stored secret config", () => { + const testDb = createTestDb(trackedDbs) + // Mirrors MAPLE_INGEST_KEY_ENCRYPTION_KEY in makeConfig() above. + const secretKey = Buffer.alloc(32, 5) + return Effect.gen(function* () { + const alerts = yield* AlertsService + const orgId = asOrgId("org_slackbot_update") + const userId = asUserId("user_slackbot_update") + + const created = yield* alerts.createDestination(orgId, userId, adminRoles, { + type: "slack-bot", + name: "Slack bot", + enabled: true, + channelId: "C111ORIG", + channelName: "alerts", + }) + assert.strictEqual(created.type, "slack-bot") + assert.strictEqual(created.summary, "#alerts") + + const readSecret = Effect.fn(function* () { + const row = yield* Effect.promise(() => + queryFirstRow<{ secret_ciphertext: string; secret_iv: string; secret_tag: string }>( + testDb, + "select secret_ciphertext, secret_iv, secret_tag from alert_destinations where id = $1", + [created.id], + ), + ) + assert.isDefined(row) + const json = yield* decryptAes256Gcm( + { ciphertext: row!.secret_ciphertext, iv: row!.secret_iv, tag: row!.secret_tag }, + secretKey, + (message) => new Error(message), + ) + return JSON.parse(json) as { type: string; channelId: string; channelName: string | null } + }) + + // A name-only update (channelId + channelName undefined) keeps both + // stored channel fields untouched. + const renamed = yield* alerts.updateDestination(orgId, userId, adminRoles, created.id, { + type: "slack-bot", + name: "Renamed bot", + }) + assert.strictEqual(renamed.name, "Renamed bot") + assert.strictEqual(renamed.summary, "#alerts") + assert.deepStrictEqual(yield* readSecret(), { + type: "slack-bot", + channelId: "C111ORIG", + channelName: "alerts", + }) + + // A provided channelName replaces the stored one; the omitted channelId + // survives. + const relabeled = yield* alerts.updateDestination(orgId, userId, adminRoles, created.id, { + type: "slack-bot", + channelName: "incidents", + }) + assert.strictEqual(relabeled.summary, "#incidents") + assert.strictEqual(relabeled.channelLabel, "#incidents") + assert.deepStrictEqual(yield* readSecret(), { + type: "slack-bot", + channelId: "C111ORIG", + channelName: "incidents", + }) + + // A blank channelId falls back to the stored value instead of wiping it. + yield* alerts.updateDestination(orgId, userId, adminRoles, created.id, { + type: "slack-bot", + channelId: " ", + }) + assert.deepStrictEqual(yield* readSecret(), { + type: "slack-bot", + channelId: "C111ORIG", + channelName: "incidents", + }) + + // A real channelId replaces the stored one; channelName stays. + yield* alerts.updateDestination(orgId, userId, adminRoles, created.id, { + type: "slack-bot", + channelId: "C222NEXT", + }) + assert.deepStrictEqual(yield* readSecret(), { + type: "slack-bot", + channelId: "C222NEXT", + channelName: "incidents", + }) + }).pipe( + Effect.provide( + makeLayer(testDb, makeWarehouseStub({ tracesAggregateRows: emptyWarehouseRows })), + ), + ) + }) + it.effect("creates an email destination and delivers a test email to every member", () => { const testDb = createTestDb(trackedDbs) const sent: Array<{ to: string; subject: string; html: string }> = [] diff --git a/apps/api/src/services/AlertsService.ts b/apps/api/src/services/AlertsService.ts index 255f1af0..b7892477 100644 --- a/apps/api/src/services/AlertsService.ts +++ b/apps/api/src/services/AlertsService.ts @@ -138,6 +138,7 @@ import { type DestinationSecretConfig, type EnrichedDestinationSecretConfig, } from "./AlertDestinationHydration" +import { resolveSlackBotTokenForDispatch } from "./SlackIntegrationService" import { dateToMs } from "../lib/time" interface NormalizedRule { @@ -579,6 +580,10 @@ const buildPublicConfig = ( summary: r.channelLabel?.trim() || "Slack incoming webhook", channelLabel: normalizeOptionalString(r.channelLabel), }), + "slack-bot": (r) => ({ + summary: r.channelName?.trim() ? `#${r.channelName.trim()}` : "Slack channel", + channelLabel: r.channelName?.trim() ? `#${r.channelName.trim()}` : null, + }), pagerduty: () => ({ summary: "PagerDuty Events API v2" as string, channelLabel: null, @@ -619,6 +624,11 @@ const buildSecretConfig = ( type: "slack" as const, webhookUrl: r.webhookUrl.trim(), }), + "slack-bot": (r) => ({ + type: "slack-bot" as const, + channelId: r.channelId.trim(), + channelName: normalizeOptionalString(r.channelName) ?? null, + }), pagerduty: (r) => ({ type: "pagerduty" as const, integrationKey: r.integrationKey.trim(), @@ -1789,7 +1799,11 @@ export class AlertsService extends Context.Service + resolveSlackBotTokenForDispatch(database, encryptionKey, orgId), + }, ) const buildPayload = (context: DeliveryPayloadContext) => ({ @@ -2144,6 +2158,35 @@ export class AlertsService extends Context.Service { + const channelName = normalizeOptionalString(r.channelName) + return Effect.succeed({ + nextPublicConfig: { + summary: + channelName != null + ? `#${channelName}` + : hydrated.publicConfig.summary, + channelLabel: + channelName != null + ? `#${channelName}` + : hydrated.publicConfig.channelLabel, + } satisfies DestinationPublicConfig, + nextSecretConfig: { + type: "slack-bot" as const, + channelId: + normalizeOptionalString(r.channelId) ?? + (hydrated.secretConfig.type === "slack-bot" + ? hydrated.secretConfig.channelId + : ""), + channelName: + r.channelName === undefined + ? hydrated.secretConfig.type === "slack-bot" + ? hydrated.secretConfig.channelName + : null + : channelName ?? null, + } satisfies DestinationSecretConfig, + }) + }, pagerduty: (r) => Effect.succeed({ nextPublicConfig: hydrated.publicConfig, diff --git a/apps/api/src/services/NotificationDispatcher.ts b/apps/api/src/services/NotificationDispatcher.ts index 6b14852c..4d3053fe 100644 --- a/apps/api/src/services/NotificationDispatcher.ts +++ b/apps/api/src/services/NotificationDispatcher.ts @@ -22,6 +22,7 @@ import { type EnrichedDestinationSecretConfig, } from "./AlertDestinationHydration" import { parseBase64Aes256GcmKey } from "../lib/Crypto" +import { resolveSlackBotTokenForDispatch } from "./SlackIntegrationService" import { Database } from "../lib/DatabaseLive" import { EmailService } from "../lib/EmailService" import { Env } from "../lib/Env" @@ -187,7 +188,11 @@ const make: Effect.Effect< DELIVERY_TIMEOUT_MS, request.linkUrl, chatUrl, - { sendEmail }, + { + sendEmail, + resolveSlackBotToken: (orgId) => + resolveSlackBotTokenForDispatch(database, encryptionKey, orgId), + }, ).pipe(Effect.tapError(() => Effect.annotateCurrentSpan({ "maple.delivery.outcome": "failed" }))) yield* Effect.annotateCurrentSpan({ "maple.delivery.outcome": "delivered", diff --git a/apps/api/src/services/SlackIntegrationService.test.ts b/apps/api/src/services/SlackIntegrationService.test.ts new file mode 100644 index 00000000..cb8eed51 --- /dev/null +++ b/apps/api/src/services/SlackIntegrationService.test.ts @@ -0,0 +1,837 @@ +import { createCipheriv, randomBytes } from "node:crypto" +import { afterEach, assert, describe, it } from "@effect/vitest" +import { ConfigProvider, Effect, Layer, Schema } from "effect" +import { TestClock } from "effect/testing" +import { FetchHttpClient } from "effect/unstable/http" +import { OrgId, UserId } from "@maple/domain/http" +import { Env } from "../lib/Env" +import { + resolveSlackBotTokenForDispatch, + SlackIntegrationService, +} from "./SlackIntegrationService" +import { ApiKeysService } from "./ApiKeysService" +import { OAuthStateRepository } from "./OAuthStateRepository" +import { Database } from "../lib/DatabaseLive" +import { cleanupTestDbs, createTestDb, executeSql, queryFirstRow, type TestDb } from "../lib/test-pglite" + +const ENCRYPTION_KEY = Buffer.alloc(32, 7) +const ENCRYPTION_KEY_B64 = ENCRYPTION_KEY.toString("base64") + +/** AES-256-GCM encrypt matching Crypto.ts's format (12-byte iv, base64 fields). */ +const encryptField = (plaintext: string, key: Buffer) => { + const iv = randomBytes(12) + const cipher = createCipheriv("aes-256-gcm", key, iv) + const ciphertext = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()]) + return { + ciphertext: ciphertext.toString("base64"), + iv: iv.toString("base64"), + tag: cipher.getAuthTag().toString("base64"), + } +} + +const makeConfig = (slackConfigured = true) => + ConfigProvider.layer( + ConfigProvider.fromUnknown({ + PORT: "3472", + TINYBIRD_HOST: "https://api.tinybird.co", + TINYBIRD_TOKEN: "test-token", + MAPLE_AUTH_MODE: "self_hosted", + MAPLE_ROOT_PASSWORD: "test-root-password", + MAPLE_DEFAULT_ORG_ID: "default", + MAPLE_INGEST_KEY_ENCRYPTION_KEY: ENCRYPTION_KEY_B64, + MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: "maple-test-lookup-secret", + MAPLE_APP_BASE_URL: "https://web.localhost", + ...(slackConfigured + ? { SLACK_CLIENT_ID: "123.abc", SLACK_CLIENT_SECRET: "shhh" } + : {}), + }), + ) + +const makeLayer = ( + testDb: TestDb, + slackConfigured = true, + apiKeysLayer: typeof ApiKeysService.layer = ApiKeysService.layer, +) => + SlackIntegrationService.layer.pipe( + Layer.provide(Layer.mergeAll(apiKeysLayer, OAuthStateRepository.layer)), + Layer.provide(testDb.layer), + Layer.provide(Env.layer), + Layer.provide(makeConfig(slackConfigured)), + ) + +/** Mirror of the service's (unexported) `SLACK_STATE_TTL_MS` — 10 minutes. */ +const SLACK_STATE_TTL_MS = 10 * 60_000 + +/** + * Wrap the real ApiKeysService so `create` runs `inject` first. `apiKeys.create` + * is the only seam between completeInstall's cross-org pre-check and its + * transaction, so this lets a test interleave a "concurrent" write there and + * exercise the transactional (setWhere) race path. + */ +const apiKeysWithInjectedCreate = (inject: () => Promise) => + Layer.effect( + ApiKeysService, + Effect.gen(function* () { + const real = yield* ApiKeysService + const create: typeof real.create = (...args) => + Effect.promise(inject).pipe(Effect.flatMap(() => real.create(...args))) + return { ...real, create } as typeof real + }), + ).pipe(Layer.provide(ApiKeysService.layer)) as typeof ApiKeysService.layer + +/** The pure dispatch helper needs only Database — build a minimal layer for it. */ +const databaseLayer = (testDb: TestDb) => testDb.layer + +const asOrgId = Schema.decodeUnknownSync(OrgId) +const asUserId = Schema.decodeUnknownSync(UserId) + +const jsonResponse = (body: unknown, status = 200) => + new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } }) + +/** A mocked `fetch` that answers Slack's `oauth.v2.access` with the current team. */ +const slackOAuthFetch = (teamRef: { current: { id: string; name: string } }): typeof globalThis.fetch => + ((input: RequestInfo | URL) => { + const url = String(input) + if (url.startsWith("https://slack.com/api/oauth.v2.access")) { + return Promise.resolve( + jsonResponse({ + ok: true, + access_token: `xoxb-${teamRef.current.id}`, + token_type: "bot", + scope: "chat:write", + bot_user_id: "U0BOT", + team: teamRef.current, + }), + ) + } + return Promise.reject(new Error(`unexpected fetch: ${url}`)) + }) as typeof globalThis.fetch + +const OAUTH_URL = "https://slack.com/api/oauth.v2.access" +const CONVERSATIONS_URL = "https://slack.com/api/conversations.list" + +/** A fetch stub scoped to one Slack endpoint; anything else rejects. */ +const slackApiFetch = ( + prefix: string, + respond: (url: string, call: number) => Response | Promise, +): typeof globalThis.fetch => { + let call = 0 + return ((input: RequestInfo | URL) => { + const url = String(input) + if (!url.startsWith(prefix)) return Promise.reject(new Error(`unexpected fetch: ${url}`)) + return Promise.resolve(respond(url, call++)) + }) as typeof globalThis.fetch +} + +const withFetch = ( + testDb: TestDb, + fetchImpl: typeof globalThis.fetch, + apiKeysLayer?: typeof ApiKeysService.layer, +) => Layer.mergeAll(makeLayer(testDb, true, apiKeysLayer), Layer.succeed(FetchHttpClient.Fetch, fetchImpl)) + +const stateFromInstallUrl = (url: string): string => { + const state = new URL(url).searchParams.get("state") + if (!state) throw new Error("install url missing state") + return state +} + +const trackedDbs: TestDb[] = [] +afterEach(() => cleanupTestDbs(trackedDbs)) + +/** Insert an active, encrypted slack_workspaces row directly (bypasses OAuth). */ +const insertWorkspace = async ( + testDb: TestDb, + opts: { id: string; orgId: string; teamId: string; teamName: string; botToken: string; apiKey: string }, +) => { + const bot = encryptField(opts.botToken, ENCRYPTION_KEY) + const key = encryptField(opts.apiKey, ENCRYPTION_KEY) + await executeSql( + testDb, + `INSERT INTO slack_workspaces ( + id, org_id, team_id, team_name, bot_user_id, scope, + bot_token_ciphertext, bot_token_iv, bot_token_tag, + api_key_id, api_key_secret_ciphertext, api_key_secret_iv, api_key_secret_tag, + installed_by_user_id, created_at, updated_at, revoked_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14, now(), now(), NULL)`, + [ + opts.id, + opts.orgId, + opts.teamId, + opts.teamName, + "U0BOT", + "chat:write", + bot.ciphertext, + bot.iv, + bot.tag, + "11111111-2222-4333-8444-555555555555", + key.ciphertext, + key.iv, + key.tag, + "user_installer", + ], + ) +} + +describe("SlackIntegrationService", () => { + it.effect("startInstall persists a single-use state and returns a Slack authorize URL", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const result = yield* slack.startInstall( + asOrgId("org_a"), + asUserId("user_a"), + "https://api.localhost/oauth/slack/callback", + ) + assert.isTrue(result.url.startsWith("https://slack.com/oauth/v2/authorize?")) + const parsed = new URL(result.url) + assert.strictEqual(parsed.searchParams.get("client_id"), "123.abc") + assert.include(parsed.searchParams.get("scope") ?? "", "chat:write") + const state = parsed.searchParams.get("state") + assert.isString(state) + + const row = yield* Effect.promise(() => + queryFirstRow<{ org_id: string; provider: string }>( + testDb, + "SELECT org_id, provider FROM oauth_auth_states WHERE state = $1", + [state], + ), + ) + assert.strictEqual(row?.org_id, "org_a") + assert.strictEqual(row?.provider, "slack") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("startInstall fails when Slack is not configured", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip( + slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb"), + ) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsValidationError") + }).pipe(Effect.provide(makeLayer(testDb, false))) + }) + + it.effect("completeInstall rejects an unknown state", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.completeInstall("code_1", "nonexistent-state")) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsValidationError") + assert.include(error.message, "not recognized") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("completeInstall rejects a state past SLACK_STATE_TTL_MS (and burns it)", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + // Create the state through the real flow at the frozen TestClock time, + // then advance the clock just past the TTL — the service reads + // Clock.currentTimeMillis, so this verifies the actual expiry window. + const start = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + const state = stateFromInstallUrl(start.url) + yield* TestClock.adjust(SLACK_STATE_TTL_MS + 1) + const error = yield* Effect.flip(slack.completeInstall("code_1", state)) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsValidationError") + assert.include(error.message, "expired") + const remaining = yield* Effect.promise(() => + queryFirstRow(testDb, "SELECT state FROM oauth_auth_states WHERE state = $1", [state]), + ) + assert.isUndefined(remaining) + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("getStatus reports not-installed for an org with no workspace", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const status = yield* slack.getStatus(asOrgId("org_none")) + assert.strictEqual(status.installed, false) + assert.isNull(status.teamId) + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("getStatus + resolveForBot read back an installed workspace", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_1", + orgId: "org_a", + teamId: "T0123", + teamName: "Acme", + botToken: "xoxb-secret-token", + apiKey: "maple_ak_secret", + }), + ) + const slack = yield* SlackIntegrationService + + const status = yield* slack.getStatus(asOrgId("org_a")) + assert.strictEqual(status.installed, true) + assert.strictEqual(status.teamId, "T0123") + assert.strictEqual(status.teamName, "Acme") + + const resolution = yield* slack.resolveForBot("T0123") + assert.strictEqual(resolution.orgId, "org_a") + assert.strictEqual(resolution.botToken, "xoxb-secret-token") + assert.strictEqual(resolution.mapleApiKey, "maple_ak_secret") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("resolveForBot fails for an unknown team", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.resolveForBot("T-unknown")) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsNotConnectedError") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("uninstall revokes the workspace so it reads as not-installed", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_2", + orgId: "org_b", + teamId: "T0999", + teamName: "Beta", + botToken: "xoxb-b", + apiKey: "maple_ak_b", + }), + ) + const slack = yield* SlackIntegrationService + const result = yield* slack.uninstall(asOrgId("org_b")) + assert.strictEqual(result.uninstalled, true) + + const status = yield* slack.getStatus(asOrgId("org_b")) + assert.strictEqual(status.installed, false) + // resolveForBot skips revoked rows too + const error = yield* Effect.flip(slack.resolveForBot("T0999")) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsNotConnectedError") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("resolveSlackBotTokenForDispatch decrypts the active workspace bot token", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_3", + orgId: "org_c", + teamId: "T0777", + teamName: "Gamma", + botToken: "xoxb-dispatch-token", + apiKey: "maple_ak_c", + }), + ) + const database = yield* Database + const token = yield* resolveSlackBotTokenForDispatch(database, ENCRYPTION_KEY, "org_c") + assert.strictEqual(token, "xoxb-dispatch-token") + }).pipe(Effect.provide(databaseLayer(testDb))) + }) + + it.effect("resolveSlackBotTokenForDispatch fails when no active install exists", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const database = yield* Database + const error = yield* Effect.flip( + resolveSlackBotTokenForDispatch(database, ENCRYPTION_KEY, "org_missing"), + ) + assert.strictEqual(error.destinationType, "slack-bot") + assert.include(error.message, "not connected") + }).pipe(Effect.provide(databaseLayer(testDb))) + }) + + it.effect("a same-org install of a second workspace replaces (revokes) the first", () => { + const testDb = createTestDb(trackedDbs) + const teamRef = { current: { id: "T1", name: "TeamOne" } } + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + + // First install → workspace T1 becomes active. + const start1 = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + yield* slack.completeInstall("code_1", stateFromInstallUrl(start1.url)) + const firstKey = yield* Effect.promise(() => + queryFirstRow<{ api_key_id: string }>( + testDb, + "SELECT api_key_id FROM slack_workspaces WHERE team_id = 'T1'", + ), + ) + assert.isString(firstKey?.api_key_id) + + // Second install of a DIFFERENT team on the SAME org → replaces T1. + teamRef.current = { id: "T2", name: "TeamTwo" } + const start2 = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + yield* slack.completeInstall("code_2", stateFromInstallUrl(start2.url)) + + // Status + dispatch resolve the NEW workspace, and exactly one row is active. + const status = yield* slack.getStatus(asOrgId("org_a")) + assert.strictEqual(status.teamId, "T2") + + const activeCount = yield* Effect.promise(() => + queryFirstRow<{ n: number }>( + testDb, + "SELECT count(*)::int AS n FROM slack_workspaces WHERE org_id = 'org_a' AND revoked_at IS NULL", + ), + ) + assert.strictEqual(activeCount?.n, 1) + + const t1Row = yield* Effect.promise(() => + queryFirstRow<{ revoked_at: string | null }>( + testDb, + "SELECT revoked_at FROM slack_workspaces WHERE team_id = 'T1'", + ), + ) + assert.isNotNull(t1Row?.revoked_at) + + const database = yield* Database + const token = yield* resolveSlackBotTokenForDispatch(database, ENCRYPTION_KEY, "org_a") + assert.strictEqual(token, "xoxb-T2") + + const t1Resolve = yield* Effect.flip(slack.resolveForBot("T1")) + assert.strictEqual(t1Resolve._tag, "@maple/http/errors/IntegrationsNotConnectedError") + + // The first workspace's minted API key was revoked. + const keyRow = yield* Effect.promise(() => + queryFirstRow<{ revoked: boolean }>(testDb, "SELECT revoked FROM api_keys WHERE id = $1", [ + firstKey!.api_key_id, + ]), + ) + assert.strictEqual(keyRow?.revoked, true) + }).pipe( + Effect.provide( + Layer.mergeAll( + makeLayer(testDb), + testDb.layer, + Layer.succeed(FetchHttpClient.Fetch, slackOAuthFetch(teamRef)), + ), + ), + ) + }) + + it.effect("the partial unique index rejects a second active row for the same org", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_a1", + orgId: "org_z", + teamId: "TX", + teamName: "X", + botToken: "b1", + apiKey: "k1", + }), + ) + let rejected = false + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_a2", + orgId: "org_z", + teamId: "TY", + teamName: "Y", + botToken: "b2", + apiKey: "k2", + }).then( + () => {}, + () => { + rejected = true + }, + ), + ) + assert.isTrue(rejected, "second active row for the same org should violate the unique index") + }).pipe(Effect.provide(databaseLayer(testDb))) + }) + + // --- Cross-org rebind rejection ----------------------------------------- + + it.effect("completeInstall rejects binding a team actively installed on another org (pre-check)", () => { + const testDb = createTestDb(trackedDbs) + const teamRef = { current: { id: "T-shared", name: "Shared" } } + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + + // org_a installs T-shared. + const s1 = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + yield* slack.completeInstall("code_1", stateFromInstallUrl(s1.url)) + + // org_b tries to install the SAME team → forbidden. + const s2 = yield* slack.startInstall(asOrgId("org_b"), asUserId("user_b"), "https://cb") + const error = yield* Effect.flip(slack.completeInstall("code_2", stateFromInstallUrl(s2.url))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsForbiddenError") + assert.include(error.message, "different Maple organization") + + // org_a's binding is untouched... + const bound = yield* Effect.promise(() => + queryFirstRow<{ org_id: string; revoked_at: string | null }>( + testDb, + "SELECT org_id, revoked_at FROM slack_workspaces WHERE team_id = 'T-shared'", + ), + ) + assert.strictEqual(bound?.org_id, "org_a") + assert.isNull(bound?.revoked_at) + + // ...and org_b ends up with no active workspace at all. + const orgBActive = yield* Effect.promise(() => + queryFirstRow<{ n: number }>( + testDb, + "SELECT count(*)::int AS n FROM slack_workspaces WHERE org_id = 'org_b' AND revoked_at IS NULL", + ), + ) + assert.strictEqual(orgBActive?.n, 0) + const status = yield* slack.getStatus(asOrgId("org_b")) + assert.strictEqual(status.installed, false) + }).pipe(Effect.provide(withFetch(testDb, slackOAuthFetch(teamRef)))) + }) + + it.effect("completeInstall rejects a cross-org rebind that lands after the pre-check (transactional race)", () => { + const testDb = createTestDb(trackedDbs) + const teamRef = { current: { id: "T-race", name: "Race" } } + // Injected between the pre-check and the transaction (apiKeys.create is the + // only seam in between): a "concurrent" install binds T-race to org_victim. + let injected = false + const inject = async () => { + if (injected) return + injected = true + await insertWorkspace(testDb, { + id: "sw_victim", + orgId: "org_victim", + teamId: "T-race", + teamName: "Victim", + botToken: "xoxb-victim", + apiKey: "maple_ak_victim", + }) + } + return Effect.gen(function* () { + // org_b already has an active workspace for a different team — the + // aborted transaction must roll back the revoke-others step for it. + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_b_old", + orgId: "org_b", + teamId: "T-old", + teamName: "Old", + botToken: "xoxb-old", + apiKey: "maple_ak_old", + }), + ) + const slack = yield* SlackIntegrationService + const start = yield* slack.startInstall(asOrgId("org_b"), asUserId("user_b"), "https://cb") + const error = yield* Effect.flip(slack.completeInstall("code_r", stateFromInstallUrl(start.url))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsForbiddenError") + assert.isTrue(injected, "the conflicting row must have been injected mid-install") + + // The victim org's binding is unchanged. + const victim = yield* Effect.promise(() => + queryFirstRow<{ org_id: string; revoked_at: string | null }>( + testDb, + "SELECT org_id, revoked_at FROM slack_workspaces WHERE team_id = 'T-race'", + ), + ) + assert.strictEqual(victim?.org_id, "org_victim") + assert.isNull(victim?.revoked_at) + + // org_b's prior workspace is still active — the in-transaction + // revocation rolled back with the aborted upsert. + const oldRow = yield* Effect.promise(() => + queryFirstRow<{ revoked_at: string | null }>( + testDb, + "SELECT revoked_at FROM slack_workspaces WHERE team_id = 'T-old'", + ), + ) + assert.isNull(oldRow?.revoked_at) + }).pipe(Effect.provide(withFetch(testDb, slackOAuthFetch(teamRef), apiKeysWithInjectedCreate(inject)))) + }) + + // --- exchangeCode failure paths (via completeInstall) --------------------- + + it.effect("completeInstall surfaces Slack's ok:false as a validation error carrying Slack's error string", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const start = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + const state = stateFromInstallUrl(start.url) + // Boundary check: at exactly the TTL the state is still valid, so the + // flow proceeds past expiry and into the (failing) code exchange. + yield* TestClock.adjust(SLACK_STATE_TTL_MS) + const error = yield* Effect.flip(slack.completeInstall("code_bad", state)) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsValidationError") + assert.include(error.message, "invalid_code") + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch(OAUTH_URL, () => jsonResponse({ ok: false, error: "invalid_code" })), + ), + ), + ) + }) + + it.effect("completeInstall maps a non-JSON OAuth response to an upstream error", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const start = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + const error = yield* Effect.flip(slack.completeInstall("code_1", stateFromInstallUrl(start.url))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError") + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch( + OAUTH_URL, + () => + new Response("maintenance", { + status: 200, + headers: { "content-type": "text/html" }, + }), + ), + ), + ), + ) + }) + + it.effect("completeInstall maps an undecodable OAuth payload to an upstream error", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const start = yield* slack.startInstall(asOrgId("org_a"), asUserId("user_a"), "https://cb") + const error = yield* Effect.flip(slack.completeInstall("code_1", stateFromInstallUrl(start.url))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError") + }).pipe( + // `ok` must be a boolean — a JSON payload with the wrong shape fails decode. + Effect.provide( + withFetch(testDb, slackApiFetch(OAUTH_URL, () => jsonResponse({ ok: "yes", team: 42 }))), + ), + ) + }) + + // --- listChannels ---------------------------------------------------------- + + it.effect("listChannels fails not-connected when the org has no active workspace", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.listChannels(asOrgId("org_none"))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsNotConnectedError") + }).pipe(Effect.provide(makeLayer(testDb))) + }) + + it.effect("listChannels returns a single page with field defaulting", () => { + const testDb = createTestDb(trackedDbs) + const urls: string[] = [] + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc1", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const channels = yield* slack.listChannels(asOrgId("org_lc")) + assert.deepStrictEqual(channels, [ + { id: "C1", name: "general", isPrivate: true, isMember: true }, + // Missing name/is_private/is_member default to id/false/false. + { id: "C2", name: "C2", isPrivate: false, isMember: false }, + ]) + assert.strictEqual(urls.length, 1) + const first = new URL(urls[0]!) + assert.isNull(first.searchParams.get("cursor")) + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch(CONVERSATIONS_URL, (url) => { + urls.push(url) + return jsonResponse({ + ok: true, + channels: [ + { id: "C1", name: "general", is_private: true, is_member: true }, + { id: "C2" }, + ], + }) + }), + ), + ), + ) + }) + + it.effect("listChannels follows next_cursor across pages and stops on an empty cursor", () => { + const testDb = createTestDb(trackedDbs) + const urls: string[] = [] + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc2", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const channels = yield* slack.listChannels(asOrgId("org_lc")) + assert.deepStrictEqual( + channels.map((c) => c.id), + ["C1", "C2"], + ) + assert.strictEqual(urls.length, 2) + assert.isNull(new URL(urls[0]!).searchParams.get("cursor")) + assert.strictEqual(new URL(urls[1]!).searchParams.get("cursor"), "cursor-2") + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch(CONVERSATIONS_URL, (url, call) => { + urls.push(url) + return call === 0 + ? jsonResponse({ + ok: true, + channels: [{ id: "C1", name: "one" }], + response_metadata: { next_cursor: "cursor-2" }, + }) + : // Slack signals "done" with an empty next_cursor. + jsonResponse({ + ok: true, + channels: [{ id: "C2", name: "two" }], + response_metadata: { next_cursor: "" }, + }) + }), + ), + ), + ) + }) + + it.effect("listChannels caps pagination at SLACK_MAX_CHANNEL_PAGES pages", () => { + const testDb = createTestDb(trackedDbs) + let calls = 0 + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc3", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const channels = yield* slack.listChannels(asOrgId("org_lc")) + // The mock ALWAYS hands back a next_cursor — the walk must stop at the + // page cap (SLACK_MAX_CHANNEL_PAGES = 3) instead of looping forever. + assert.strictEqual(calls, 3) + assert.deepStrictEqual( + channels.map((c) => c.id), + ["C-page-0", "C-page-1", "C-page-2"], + ) + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch(CONVERSATIONS_URL, (_url, call) => { + calls++ + return jsonResponse({ + ok: true, + channels: [{ id: `C-page-${call}` }], + response_metadata: { next_cursor: "more" }, + }) + }), + ), + ), + ) + }) + + it.effect("listChannels maps Slack ok:false to an upstream error", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc4", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.listChannels(asOrgId("org_lc"))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError") + assert.include(error.message, "invalid_auth") + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch(CONVERSATIONS_URL, () => jsonResponse({ ok: false, error: "invalid_auth" })), + ), + ), + ) + }) + + it.effect("listChannels maps a non-JSON response to an upstream error", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc5", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.listChannels(asOrgId("org_lc"))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError") + }).pipe( + Effect.provide( + withFetch( + testDb, + slackApiFetch( + CONVERSATIONS_URL, + () => + new Response("gateway timeout", { + status: 200, + headers: { "content-type": "text/plain" }, + }), + ), + ), + ), + ) + }) + + it.effect("listChannels maps an undecodable payload to an upstream error", () => { + const testDb = createTestDb(trackedDbs) + return Effect.gen(function* () { + yield* Effect.promise(() => + insertWorkspace(testDb, { + id: "sw_lc6", + orgId: "org_lc", + teamId: "T-LC", + teamName: "LC", + botToken: "xoxb-lc-token", + apiKey: "maple_ak_lc", + }), + ) + const slack = yield* SlackIntegrationService + const error = yield* Effect.flip(slack.listChannels(asOrgId("org_lc"))) + assert.strictEqual(error._tag, "@maple/http/errors/IntegrationsUpstreamError") + }).pipe( + // channels must be an array of objects with a string id. + Effect.provide( + withFetch( + testDb, + slackApiFetch(CONVERSATIONS_URL, () => jsonResponse({ ok: true, channels: [{ id: 42 }] })), + ), + ), + ) + }) +}) diff --git a/apps/api/src/services/SlackIntegrationService.ts b/apps/api/src/services/SlackIntegrationService.ts new file mode 100644 index 00000000..eee0fee2 --- /dev/null +++ b/apps/api/src/services/SlackIntegrationService.ts @@ -0,0 +1,813 @@ +import { randomBytes, randomUUID } from "node:crypto" +import { + AlertDeliveryError, + ApiKeyId, + IntegrationsForbiddenError, + IntegrationsNotConnectedError, + IntegrationsPersistenceError, + IntegrationsUpstreamError, + IntegrationsValidationError, + OrgId, + UserId, +} from "@maple/domain/http" +import { slackWorkspaces, type SlackWorkspaceRow } from "@maple/db" +import { and, eq, isNotNull, isNull, ne, or } from "drizzle-orm" +import { Clock, Context, Effect, Layer, Option, Redacted, Schema } from "effect" +import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/unstable/http" +import { + decryptAes256Gcm, + encryptAes256Gcm, + parseBase64Aes256GcmKey, + type EncryptedValue, +} from "../lib/Crypto" +import { Database, type DatabaseShape } from "../lib/DatabaseLive" +import { Env } from "../lib/Env" +import { ApiKeysService } from "./ApiKeysService" +import { OAuthStateRepository } from "./OAuthStateRepository" + +const SLACK_PROVIDER = "slack" +const SLACK_STATE_TTL_MS = 10 * 60_000 // 10 minutes +const SLACK_OAUTH_ACCESS_URL = "https://slack.com/api/oauth.v2.access" +const SLACK_CONVERSATIONS_LIST_URL = "https://slack.com/api/conversations.list" +const SLACK_MAX_CHANNEL_PAGES = 3 + +/** Public callback path Slack redirects to after an install (mounted in app.ts). */ +export const SLACK_CALLBACK_PATH = "/oauth/slack/callback" + +/** + * The bot scopes Maple requests at install time. `chat:write.public` lets the + * bot post to public channels it hasn't been invited to; `im:*` power the DM + * agent surface. Keep in sync with the Slack app manifest. + */ +export const SLACK_BOT_SCOPES = [ + "app_mentions:read", + "chat:write", + "chat:write.public", + "channels:read", + "groups:read", + "im:history", + "im:read", + "im:write", + "users:read", +].join(",") + +/** + * Thrown inside the install transaction (throwing is the only way to make a + * drizzle transaction roll back) when the same-team upsert is blocked by an + * active binding owned by another org. Caught by `instanceof` immediately + * around the transaction and converted into a discriminated result — it never + * escapes `completeInstall`. + */ +class SlackCrossOrgConflict extends Error { + constructor() { + super("Slack workspace is actively bound to a different organization") + this.name = "SlackCrossOrgConflict" + } +} + +const decodeApiKeyIdOption = Schema.decodeUnknownOption(ApiKeyId) +const decodeOrgId = Schema.decodeUnknownEffect(OrgId) +const decodeUserId = Schema.decodeUnknownEffect(UserId) + +// --- Slack API response shapes --------------------------------------------- + +const SlackOAuthAccessSchema = Schema.Struct({ + ok: Schema.Boolean, + error: Schema.optionalKey(Schema.String), + access_token: Schema.optionalKey(Schema.String), + token_type: Schema.optionalKey(Schema.String), + scope: Schema.optionalKey(Schema.String), + bot_user_id: Schema.optionalKey(Schema.String), + team: Schema.optionalKey( + Schema.Struct({ + id: Schema.String, + name: Schema.optionalKey(Schema.String), + }), + ), +}) +const decodeOAuthAccess = Schema.decodeUnknownEffect(SlackOAuthAccessSchema) + +const SlackConversationsListSchema = Schema.Struct({ + ok: Schema.Boolean, + error: Schema.optionalKey(Schema.String), + channels: Schema.optionalKey( + Schema.Array( + Schema.Struct({ + id: Schema.String, + name: Schema.optionalKey(Schema.String), + is_private: Schema.optionalKey(Schema.Boolean), + is_member: Schema.optionalKey(Schema.Boolean), + }), + ), + ), + response_metadata: Schema.optionalKey( + Schema.Struct({ next_cursor: Schema.optionalKey(Schema.String) }), + ), +}) +const decodeConversationsList = Schema.decodeUnknownEffect(SlackConversationsListSchema) + +// --- Public types ----------------------------------------------------------- + +export interface SlackInstallStatus { + readonly installed: boolean + readonly teamId: string | null + readonly teamName: string | null + readonly botUserId: string | null + readonly installedAt: number | null +} + +export interface SlackChannelSummary { + readonly id: string + readonly name: string + readonly isPrivate: boolean + readonly isMember: boolean +} + +export interface SlackBotResolution { + readonly orgId: string + readonly teamId: string + readonly teamName: string | null + readonly botToken: string + readonly mapleApiKey: string +} + +/** + * Wire schema for the FIXED internal resolve response contract + * (`GET /internal/slack/workspaces/:teamId`). The Railway-hosted bot is built + * against exactly these keys — do not rename or drop fields. + */ +export const SlackBotResolutionResponseSchema = Schema.Struct({ + orgId: Schema.String, + teamId: Schema.String, + teamName: Schema.NullOr(Schema.String), + botToken: Schema.String, + mapleApiKey: Schema.String, +}) + +// --- Shared decrypt helper (also used by the alert dispatch path) ----------- + +const loadActiveWorkspaceByOrg = Effect.fnUntraced(function* ( + database: DatabaseShape, + orgId: string, +) { + const rows = yield* database.execute((db) => + db + .select() + .from(slackWorkspaces) + .where(and(eq(slackWorkspaces.orgId, orgId), isNull(slackWorkspaces.revokedAt))) + .limit(1), + ) + return Option.fromNullishOr(rows[0]) +}) + +/** + * Resolve the decrypted Slack bot token for an org's active installation. + * Shared by the alert-delivery `slack-bot` arm (both apps/api and the alerting + * worker) so those dispatchers do not need the full SlackIntegrationService. + * Fails with an {@link AlertDeliveryError} when there is no active install. + */ +export const resolveSlackBotTokenForDispatch = Effect.fn( + "SlackIntegrationService.resolveSlackBotTokenForDispatch", +)(function* (database: DatabaseShape, encryptionKey: Buffer, orgId: string) { + yield* Effect.annotateCurrentSpan({ orgId }) + const row = yield* loadActiveWorkspaceByOrg(database, orgId).pipe( + Effect.mapError( + (error) => + new AlertDeliveryError({ + message: `Failed to load Slack installation: ${error.message}`, + destinationType: "slack-bot", + }), + ), + ) + if (Option.isNone(row)) { + return yield* Effect.fail( + new AlertDeliveryError({ + message: "Slack is not connected for this organization — install the Maple Slack app", + destinationType: "slack-bot", + }), + ) + } + return yield* decryptAes256Gcm( + { + ciphertext: row.value.botTokenCiphertext, + iv: row.value.botTokenIv, + tag: row.value.botTokenTag, + }, + encryptionKey, + () => + new AlertDeliveryError({ + message: "Failed to decrypt stored Slack bot token", + destinationType: "slack-bot", + }), + ) +}) + +// --- Service ---------------------------------------------------------------- + +export interface SlackIntegrationServiceShape { + readonly startInstall: ( + orgId: OrgId, + userId: UserId, + callbackUrl: string, + ) => Effect.Effect< + { readonly url: string }, + IntegrationsValidationError | IntegrationsPersistenceError + > + readonly completeInstall: ( + code: string, + state: string, + ) => Effect.Effect< + { readonly orgId: OrgId; readonly teamName: string | null }, + | IntegrationsValidationError + | IntegrationsForbiddenError + | IntegrationsUpstreamError + | IntegrationsPersistenceError + > + readonly getStatus: ( + orgId: OrgId, + ) => Effect.Effect + readonly uninstall: ( + orgId: OrgId, + ) => Effect.Effect<{ readonly uninstalled: boolean }, IntegrationsPersistenceError> + readonly listChannels: ( + orgId: OrgId, + ) => Effect.Effect< + ReadonlyArray, + IntegrationsNotConnectedError | IntegrationsUpstreamError | IntegrationsPersistenceError + > + readonly resolveForBot: ( + teamId: string, + ) => Effect.Effect +} + +export class SlackIntegrationService extends Context.Service< + SlackIntegrationService, + SlackIntegrationServiceShape +>()("@maple/api/services/SlackIntegrationService", { + make: Effect.gen(function* () { + const database = yield* Database + const env = yield* Env + const apiKeys = yield* ApiKeysService + const states = yield* OAuthStateRepository + const httpClient = yield* HttpClient.HttpClient + + const encryptionKey = yield* parseBase64Aes256GcmKey( + Redacted.value(env.MAPLE_INGEST_KEY_ENCRYPTION_KEY), + (message) => + new IntegrationsValidationError({ + message: + message === "Expected a non-empty base64 encryption key" + ? "MAPLE_INGEST_KEY_ENCRYPTION_KEY is required" + : message === "Expected base64 for exactly 32 bytes" + ? "MAPLE_INGEST_KEY_ENCRYPTION_KEY must be base64 for exactly 32 bytes" + : message, + }), + ) + + const toPersistenceError = (error: { readonly message: string }) => + new IntegrationsPersistenceError({ message: error.message }) + + const encryptValue = (plaintext: string): Effect.Effect => + encryptAes256Gcm( + plaintext, + encryptionKey, + (message) => new IntegrationsPersistenceError({ message: `Encryption failed: ${message}` }), + ) + + const decryptValue = ( + encrypted: EncryptedValue, + ): Effect.Effect => + decryptAes256Gcm( + encrypted, + encryptionKey, + () => new IntegrationsPersistenceError({ message: "Failed to decrypt stored Slack secret" }), + ) + + const requireOAuthClient = Effect.gen(function* () { + const clientId = Option.getOrUndefined(env.SLACK_CLIENT_ID) + const clientSecret = Option.match(env.SLACK_CLIENT_SECRET, { + onNone: () => undefined, + onSome: (value) => Redacted.value(value), + }) + if (!clientId || !clientSecret) { + return yield* Effect.fail( + new IntegrationsValidationError({ + message: "Slack integration is not configured (SLACK_CLIENT_ID / SLACK_CLIENT_SECRET)", + }), + ) + } + return { clientId, clientSecret } + }) + + const startInstall = Effect.fn("SlackIntegrationService.startInstall")(function* ( + orgId: OrgId, + userId: UserId, + callbackUrl: string, + ) { + yield* Effect.annotateCurrentSpan({ orgId }) + const { clientId } = yield* requireOAuthClient + const state = randomBytes(24).toString("base64url") + const now = yield* Clock.currentTimeMillis + yield* states.purgeExpired(now).pipe(Effect.mapError(toPersistenceError)) + yield* states + .insert({ + state, + orgId, + provider: SLACK_PROVIDER, + initiatedByUserId: userId, + redirectUri: callbackUrl, + returnTo: null, + createdAt: new Date(now), + expiresAt: new Date(now + SLACK_STATE_TTL_MS), + }) + .pipe(Effect.mapError(toPersistenceError)) + + const params = new URLSearchParams({ + client_id: clientId, + scope: SLACK_BOT_SCOPES, + state, + redirect_uri: callbackUrl, + }) + return { url: `https://slack.com/oauth/v2/authorize?${params.toString()}` } + }) + + const exchangeCode = Effect.fn("SlackIntegrationService.exchangeCode")(function* ( + code: string, + redirectUri: string, + clientId: string, + clientSecret: string, + ) { + const request = HttpClientRequest.post(SLACK_OAUTH_ACCESS_URL, { + headers: { accept: "application/json" }, + }).pipe( + HttpClientRequest.bodyUrlParams({ + client_id: clientId, + client_secret: clientSecret, + code, + redirect_uri: redirectUri, + }), + ) + const response = yield* httpClient.execute(request).pipe( + Effect.mapError( + (error) => new IntegrationsUpstreamError({ message: `Slack OAuth request failed: ${error.message}` }), + ), + ) + const json = yield* response.json.pipe( + Effect.mapError( + (error) => + new IntegrationsUpstreamError({ + message: `Slack OAuth returned a non-JSON response: ${error.message}`, + }), + ), + ) + const decoded = yield* decodeOAuthAccess(json).pipe( + Effect.mapError( + (error) => + new IntegrationsUpstreamError({ + message: `Slack OAuth returned an unexpected payload: ${error.message}`, + }), + ), + ) + const accessToken = decoded.access_token + const team = decoded.team + if (!decoded.ok || !accessToken || !team?.id) { + return yield* Effect.fail( + new IntegrationsValidationError({ + message: `Slack authorization failed: ${decoded.error ?? "missing token or team"}`, + }), + ) + } + return { + accessToken, + teamId: team.id, + teamName: team.name ?? null, + botUserId: decoded.bot_user_id ?? null, + scope: decoded.scope ?? null, + } + }) + + const completeInstall = Effect.fn("SlackIntegrationService.completeInstall")(function* ( + code: string, + state: string, + ) { + const { clientId, clientSecret } = yield* requireOAuthClient + const stateRow = yield* states.findByState(state).pipe(Effect.mapError(toPersistenceError)) + if (Option.isNone(stateRow) || stateRow.value.provider !== SLACK_PROVIDER) { + return yield* Effect.fail( + new IntegrationsValidationError({ + message: "Slack state not recognized — restart the install flow", + }), + ) + } + const row = stateRow.value + const now = yield* Clock.currentTimeMillis + if (row.expiresAt.getTime() < now) { + yield* states.deleteByState(state).pipe(Effect.mapError(toPersistenceError)) + return yield* Effect.fail( + new IntegrationsValidationError({ + message: "Slack state expired — restart the install flow", + }), + ) + } + // Single-use: burn the state before doing any side effects. + yield* states.deleteByState(state).pipe(Effect.mapError(toPersistenceError)) + + const orgId = yield* decodeOrgId(row.orgId).pipe( + Effect.mapError( + (error) => + new IntegrationsPersistenceError({ + message: `Stored Slack OAuth state has an invalid orgId: ${error.message}`, + }), + ), + ) + yield* Effect.annotateCurrentSpan({ orgId }) + const initiatedByUserId = yield* decodeUserId(row.initiatedByUserId).pipe( + Effect.mapError( + (error) => + new IntegrationsPersistenceError({ + message: `Stored Slack OAuth state has an invalid userId: ${error.message}`, + }), + ), + ) + const access = yield* exchangeCode(code, row.redirectUri, clientId, clientSecret) + const teamId = access.teamId + const teamName = access.teamName + + // Reject re-binding a team that is actively installed on a different org. + const existingByTeam = yield* database + .execute((db) => + db.select().from(slackWorkspaces).where(eq(slackWorkspaces.teamId, teamId)).limit(1), + ) + .pipe( + Effect.mapError(toPersistenceError), + Effect.map((rows) => Option.fromNullishOr(rows[0])), + ) + if ( + Option.isSome(existingByTeam) && + existingByTeam.value.orgId !== orgId && + existingByTeam.value.revokedAt === null + ) { + return yield* Effect.fail( + new IntegrationsForbiddenError({ + message: + "This Slack workspace is already connected to a different Maple organization. Uninstall it there first.", + }), + ) + } + const priorRow = Option.getOrUndefined(existingByTeam) + + // Mint a full-access MCP-kind API key for the bot; capture the plaintext. + const created = yield* apiKeys + .create(orgId, initiatedByUserId, { + name: `Slack bot (${teamName ?? teamId})`, + kind: "mcp", + scopes: null, + }) + .pipe( + Effect.mapError( + (error) => + new IntegrationsPersistenceError({ + message: `Failed to mint Slack API key: ${error.message}`, + }), + ), + ) + + const botTokenEnc = yield* encryptValue(access.accessToken) + const apiKeyEnc = yield* encryptValue(created.secret) + + const values = { + id: priorRow?.id ?? randomUUID(), + orgId, + teamId, + teamName, + botUserId: access.botUserId, + scope: access.scope ?? SLACK_BOT_SCOPES, + botTokenCiphertext: botTokenEnc.ciphertext, + botTokenIv: botTokenEnc.iv, + botTokenTag: botTokenEnc.tag, + apiKeyId: created.id, + apiKeySecretCiphertext: apiKeyEnc.ciphertext, + apiKeySecretIv: apiKeyEnc.iv, + apiKeySecretTag: apiKeyEnc.tag, + installedByUserId: row.initiatedByUserId, + createdAt: new Date(priorRow ? priorRow.createdAt.getTime() : now), + updatedAt: new Date(now), + revokedAt: null, + } + + // Invariant: at most one active (revoked_at IS NULL) row per org. In one + // transaction, first revoke any OTHER active workspace for this org (a + // same-org install of a different team replaces the old one), then upsert + // the new/refreshed row. The `setWhere` guard makes the cross-org rejection + // race-safe: a concurrent install of the same team by a different org can't + // clobber the existing binding — the update is skipped and we abort the + // transaction (throwing is the only way to make drizzle roll back), then + // convert the abort into a discriminated `crossOrgConflict` result. The + // partial unique index on (org_id) is the backstop that ultimately enforces + // the single-active-row invariant. + const writeResult = yield* database + .execute< + | { readonly _tag: "ok"; readonly revokedOtherKeyIds: Array } + | { readonly _tag: "crossOrgConflict" } + >(async (db) => { + try { + return await db.transaction(async (tx) => { + const revokedOthers = await tx + .update(slackWorkspaces) + .set({ revokedAt: new Date(now), updatedAt: new Date(now) }) + .where( + and( + eq(slackWorkspaces.orgId, orgId), + isNull(slackWorkspaces.revokedAt), + ne(slackWorkspaces.teamId, teamId), + ), + ) + .returning({ apiKeyId: slackWorkspaces.apiKeyId }) + + const upserted = await tx + .insert(slackWorkspaces) + .values(values) + .onConflictDoUpdate({ + target: slackWorkspaces.teamId, + // Only allow overwriting a same-team row that belongs to this + // org or has already been revoked — never an active binding + // owned by another org. + setWhere: or( + eq(slackWorkspaces.orgId, orgId), + isNotNull(slackWorkspaces.revokedAt), + ), + set: { + orgId: values.orgId, + teamName: values.teamName, + botUserId: values.botUserId, + scope: values.scope, + botTokenCiphertext: values.botTokenCiphertext, + botTokenIv: values.botTokenIv, + botTokenTag: values.botTokenTag, + apiKeyId: values.apiKeyId, + apiKeySecretCiphertext: values.apiKeySecretCiphertext, + apiKeySecretIv: values.apiKeySecretIv, + apiKeySecretTag: values.apiKeySecretTag, + installedByUserId: values.installedByUserId, + createdAt: values.createdAt, + updatedAt: values.updatedAt, + revokedAt: null, + }, + }) + .returning({ id: slackWorkspaces.id }) + + // Zero rows means the same-team conflict hit an active row owned by a + // different org (the setWhere blocked it) — abort so revoke-others + // rolls back too. + if (upserted.length === 0) throw new SlackCrossOrgConflict() + + return { + _tag: "ok" as const, + revokedOtherKeyIds: revokedOthers.map((r) => r.apiKeyId), + } + }) + } catch (error) { + if (error instanceof SlackCrossOrgConflict) { + return { _tag: "crossOrgConflict" as const } + } + throw error + } + }) + .pipe(Effect.mapError(toPersistenceError)) + if (writeResult._tag === "crossOrgConflict") { + return yield* Effect.fail( + new IntegrationsForbiddenError({ + message: + "This Slack workspace is already connected to a different Maple organization. Uninstall it there first.", + }), + ) + } + + // Best-effort: revoke the API keys of every workspace this install + // replaced — the prior same-team row (whose key we just rotated) and any + // other-team rows we deactivated above. Bookkeeping must not fail the + // install now that the DB write has committed. + const keyIdsToRevoke = [priorRow?.apiKeyId ?? null, ...writeResult.revokedOtherKeyIds] + yield* Effect.forEach(keyIdsToRevoke, (rawKeyId) => { + if (!rawKeyId) return Effect.void + const keyId = decodeApiKeyIdOption(rawKeyId) + return Option.isSome(keyId) + ? apiKeys.revoke(orgId, keyId.value).pipe(Effect.ignore) + : Effect.void + }) + + return { orgId, teamName } + }) + + const getStatus = Effect.fn("SlackIntegrationService.getStatus")(function* (orgId: OrgId) { + yield* Effect.annotateCurrentSpan({ orgId }) + const rowOption = yield* loadActiveWorkspaceByOrg(database, orgId).pipe( + Effect.mapError(toPersistenceError), + ) + return Option.match(rowOption, { + onNone: () => + ({ + installed: false, + teamId: null, + teamName: null, + botUserId: null, + installedAt: null, + }) satisfies SlackInstallStatus, + onSome: (row) => + ({ + installed: true, + teamId: row.teamId, + teamName: row.teamName, + botUserId: row.botUserId, + installedAt: row.createdAt.getTime(), + }) satisfies SlackInstallStatus, + }) + }) + + const uninstall = Effect.fn("SlackIntegrationService.uninstall")(function* (orgId: OrgId) { + yield* Effect.annotateCurrentSpan({ orgId }) + const rowOption = yield* loadActiveWorkspaceByOrg(database, orgId).pipe( + Effect.mapError(toPersistenceError), + ) + if (Option.isNone(rowOption)) return { uninstalled: false } + const row = rowOption.value + const now = yield* Clock.currentTimeMillis + // Revoke the minted API key (best-effort — bookkeeping must not fail the uninstall). + if (row.apiKeyId) { + const keyId = decodeApiKeyIdOption(row.apiKeyId) + if (Option.isSome(keyId)) { + yield* apiKeys.revoke(orgId, keyId.value).pipe(Effect.ignore) + } + } + yield* database + .execute((db) => + db + .update(slackWorkspaces) + .set({ revokedAt: new Date(now), updatedAt: new Date(now) }) + .where(eq(slackWorkspaces.id, row.id)), + ) + .pipe(Effect.mapError(toPersistenceError)) + return { uninstalled: true } + }) + + /** Fetch one `conversations.list` page; returns the page's channels + next cursor. */ + const fetchChannelPage = Effect.fnUntraced(function* ( + botToken: string, + cursor: Option.Option, + ) { + const params = new URLSearchParams({ + types: "public_channel,private_channel", + exclude_archived: "true", + limit: "200", + }) + Option.map(cursor, (value) => params.set("cursor", value)) + const response = yield* httpClient + .get(`${SLACK_CONVERSATIONS_LIST_URL}?${params.toString()}`, { + headers: { authorization: `Bearer ${botToken}`, accept: "application/json" }, + }) + .pipe( + Effect.mapError( + (error) => + new IntegrationsUpstreamError({ + message: `Slack conversations.list failed: ${error.message}`, + }), + ), + ) + const json = yield* response.json.pipe( + Effect.mapError( + (error) => + new IntegrationsUpstreamError({ + message: `Slack conversations.list returned a non-JSON response: ${error.message}`, + }), + ), + ) + const decoded = yield* decodeConversationsList(json).pipe( + Effect.mapError( + (error) => + new IntegrationsUpstreamError({ + message: `Slack conversations.list returned an unexpected payload: ${error.message}`, + }), + ), + ) + if (!decoded.ok) { + return yield* Effect.fail( + new IntegrationsUpstreamError({ + message: `Slack conversations.list error: ${decoded.error ?? "unknown"}`, + }), + ) + } + const channels: ReadonlyArray = (decoded.channels ?? []).map( + (channel) => ({ + id: channel.id, + name: channel.name ?? channel.id, + isPrivate: channel.is_private ?? false, + isMember: channel.is_member ?? false, + }), + ) + // Slack signals "no more pages" with a missing or empty next_cursor. + const next = decoded.response_metadata?.next_cursor + return { channels, next: next ? Option.some(next) : Option.none() } + }) + + /** + * Cursor-driven page walk with pure accumulation, capped at + * {@link SLACK_MAX_CHANNEL_PAGES} pages. + */ + const collectChannelPages = ( + botToken: string, + cursor: Option.Option, + page: number, + acc: ReadonlyArray, + ): Effect.Effect< + ReadonlyArray, + IntegrationsUpstreamError | IntegrationsPersistenceError + > => + page >= SLACK_MAX_CHANNEL_PAGES + ? Effect.succeed(acc) + : fetchChannelPage(botToken, cursor).pipe( + Effect.flatMap(({ channels, next }) => + Option.match(next, { + onNone: () => Effect.succeed([...acc, ...channels]), + onSome: (nextCursor) => + collectChannelPages(botToken, Option.some(nextCursor), page + 1, [ + ...acc, + ...channels, + ]), + }), + ), + ) + + const listChannels = Effect.fn("SlackIntegrationService.listChannels")(function* (orgId: OrgId) { + yield* Effect.annotateCurrentSpan({ orgId }) + const rowOption = yield* loadActiveWorkspaceByOrg(database, orgId).pipe( + Effect.mapError(toPersistenceError), + ) + if (Option.isNone(rowOption)) { + return yield* Effect.fail( + new IntegrationsNotConnectedError({ + message: "Slack is not connected for this organization", + }), + ) + } + const row = rowOption.value + const botToken = yield* decryptValue({ + ciphertext: row.botTokenCiphertext, + iv: row.botTokenIv, + tag: row.botTokenTag, + }) + return yield* collectChannelPages(botToken, Option.none(), 0, []) + }) + + const resolveForBot = Effect.fn("SlackIntegrationService.resolveForBot")(function* (teamId: string) { + yield* Effect.annotateCurrentSpan({ teamId }) + const rowOption: Option.Option = yield* database + .execute((db) => + db + .select() + .from(slackWorkspaces) + .where(and(eq(slackWorkspaces.teamId, teamId), isNull(slackWorkspaces.revokedAt))) + .limit(1), + ) + .pipe( + Effect.mapError(toPersistenceError), + Effect.map((rows) => Option.fromNullishOr(rows[0])), + ) + if (Option.isNone(rowOption)) { + return yield* Effect.fail( + new IntegrationsNotConnectedError({ + message: "No active Slack installation for this team", + }), + ) + } + const row = rowOption.value + const botToken = yield* decryptValue({ + ciphertext: row.botTokenCiphertext, + iv: row.botTokenIv, + tag: row.botTokenTag, + }) + const mapleApiKey = yield* decryptValue({ + ciphertext: row.apiKeySecretCiphertext, + iv: row.apiKeySecretIv, + tag: row.apiKeySecretTag, + }) + return { + orgId: row.orgId, + teamId: row.teamId, + teamName: row.teamName, + botToken, + mapleApiKey, + } satisfies SlackBotResolution + }) + + return { + startInstall, + completeInstall, + getStatus, + uninstall, + listChannels, + resolveForBot, + } satisfies SlackIntegrationServiceShape + }), +}) { + static readonly layer = Layer.effect(this, this.make).pipe(Layer.provide(FetchHttpClient.layer)) +} diff --git a/apps/electric-sync/src/routes/shape.http.ts b/apps/electric-sync/src/routes/shape.http.ts index 7c9080fe..0e39b2b8 100644 --- a/apps/electric-sync/src/routes/shape.http.ts +++ b/apps/electric-sync/src/routes/shape.http.ts @@ -215,6 +215,23 @@ export const buildUpstreamShapeUrl = (args: { return url.toString() } +/** + * Electric Cloud authenticates every shape request with a source `secret` + * (paired with `source_id`); self-hosted Electric (local docker) needs neither. + * A config carrying exactly ONE of the two is incoherent — most commonly a + * deploy that inherited a shared `ELECTRIC_URL` + `ELECTRIC_SOURCE_ID` from the + * secret store but no matching `ELECTRIC_SECRET` (e.g. a PR preview whose per-PR + * Electric source step was skipped because the CLI token isn't provisioned). + * Forwarding that upstream is a guaranteed 401 `MISSING_SECRET` from Electric + * Cloud, which surfaces as a hard-broken shape stream in the browser. Instead we + * treat it as "not configured" and take the same 503 graceful-degrade path as a + * missing `ELECTRIC_URL`. Pure + exported so the coherence rule is unit-tested. + */ +export const isElectricConfigCoherent = (creds: { + readonly sourceId: string | undefined + readonly secret: string | undefined +}): boolean => (creds.sourceId === undefined) === (creds.secret === undefined) + const errorText = (message: string, status: number) => HttpServerResponse.text(message, { status, @@ -233,13 +250,31 @@ export const ElectricSyncRouter = HttpRouter.use((router) => onNone: () => undefined, onSome: Redacted.value, }) + // Electric Cloud needs `source_id` + `secret` together; a half-configured + // deploy would forward a doomed unauthenticated request (401 MISSING_SECRET). + // Detect it once at isolate startup and disable sync (503 below) rather than + // per-request, and log so the misconfiguration is discoverable in telemetry. + const configCoherent = isElectricConfigCoherent({ sourceId, secret }) + if (electricUrl && !configCoherent) { + yield* Effect.logWarning( + "Electric sync disabled: incoherent Cloud credentials — set both ELECTRIC_SOURCE_ID and ELECTRIC_SECRET, or neither", + ).pipe( + Effect.annotateLogs({ + hasSourceId: sourceId !== undefined, + hasSecret: secret !== undefined, + }), + ) + } const handle = (req: HttpServerRequest.HttpServerRequest) => Effect.gen(function* () { - // Not configured (e.g. self-hosted without an Electric container) → - // 503; the web app's collections degrade and it keeps using its - // existing effect-atom fetches. - if (!electricUrl) return errorText("Electric sync is not configured", 503) + // Not configured (self-hosted without an Electric container, or a + // half-configured Cloud deploy missing its source secret) → 503; the + // web app's collections degrade and it keeps using its existing + // effect-atom fetches. + if (!electricUrl || !configCoherent) { + return errorText("Electric sync is not configured", 503) + } const requestUrl = new URL(req.url, "http://internal") const shapeParam = requestUrl.searchParams.get("shape") diff --git a/apps/electric-sync/src/routes/shape.test.ts b/apps/electric-sync/src/routes/shape.test.ts index 5bfe10cb..617dc495 100644 --- a/apps/electric-sync/src/routes/shape.test.ts +++ b/apps/electric-sync/src/routes/shape.test.ts @@ -1,5 +1,10 @@ import { assert, describe, it } from "@effect/vitest" -import { buildUpstreamShapeUrl, isShapeName, shapeResponseHeaders } from "./shape.http" +import { + buildUpstreamShapeUrl, + isElectricConfigCoherent, + isShapeName, + shapeResponseHeaders, +} from "./shape.http" const parse = (raw: string) => { const url = new URL(raw) @@ -23,6 +28,24 @@ describe("isShapeName", () => { }) }) +describe("isElectricConfigCoherent", () => { + it("accepts self-hosted (neither source id nor secret)", () => { + assert.isTrue(isElectricConfigCoherent({ sourceId: undefined, secret: undefined })) + }) + + it("accepts Electric Cloud (both source id and secret)", () => { + assert.isTrue(isElectricConfigCoherent({ sourceId: "src_1", secret: "sh_secret" })) + }) + + it("rejects a source id without a secret (the skipped-per-PR-source case)", () => { + assert.isFalse(isElectricConfigCoherent({ sourceId: "src_1", secret: undefined })) + }) + + it("rejects a secret without a source id", () => { + assert.isFalse(isElectricConfigCoherent({ sourceId: undefined, secret: "sh_secret" })) + }) +}) + describe("buildUpstreamShapeUrl", () => { const base = { electricUrl: "http://electric:3000", diff --git a/apps/slack-agent/.dockerignore b/apps/slack-agent/.dockerignore new file mode 100644 index 00000000..f3e61e1e --- /dev/null +++ b/apps/slack-agent/.dockerignore @@ -0,0 +1,8 @@ +node_modules +.eve +.output +.nitro +.env* +!.env.local.example +*.tsbuildinfo +.DS_Store diff --git a/apps/slack-agent/.env.local.example b/apps/slack-agent/.env.local.example new file mode 100644 index 00000000..1f6a8e14 --- /dev/null +++ b/apps/slack-agent/.env.local.example @@ -0,0 +1,62 @@ +# Copy to .env.local (gitignored) for local dev, and set the same values as +# service variables on Railway. eve auto-loads .env.local from the app root. + +# ── Model: Cloudflare Workers AI (REST) ───────────────────────────────────── +# Dashboard → AI → Workers AI, and an API token scoped to "Workers AI". +CLOUDFLARE_ACCOUNT_ID= +CLOUDFLARE_API_TOKEN= +# Optional overrides (defaults live in agent/agent.ts): +# WORKERS_AI_MODEL=@cf/zai-org/glm-5.2 # must stream structured tool calls — see README Notes +# WORKERS_AI_CONTEXT_WINDOW=262144 + +# ── Slack (self-managed, multi-workspace app — no Vercel Connect) ─────────── +# Signing secret is per-app/static: it HMAC-verifies every inbound webhook. +# From your Slack app: Basic Information → Signing Secret. +SLACK_SIGNING_SECRET= +# SLACK_BOT_TOKEN is the single-workspace DEV fallback only. In multi-workspace +# mode the bot token is resolved per team from the Maple API (below); leave this +# unset in production. Set it to a xoxb-... token to run against one workspace +# without a Maple mapping (e.g. local iteration on the model + tool loop). +# SLACK_BOT_TOKEN= + +# ── Maple API (multi-workspace resolve + MCP) ─────────────────────────────── +# Per-team installs live in Maple's API. The agent resolves each Slack team's +# bot token + Maple API key from an internal endpoint, and connects to Maple's +# MCP server for observability tools. +# +# NOTE: MAPLE_API_BASE_URL is used to build the MCP connection URL, which eve +# BAKES INTO its compiled manifest at build time (like EVE_WORKFLOW_WORLD). It +# must be set when `eve build` runs for the production URL to bake correctly — a +# runtime-only value is too late for the connection URL. The Dockerfile passes it +# at build. Unset locally → defaults to https://api.localhost. +# MAPLE_API_BASE_URL=https://api.maple.dev +# Internal service token (the code sends it as `Bearer maple_svc_`). +MAPLE_INTERNAL_SERVICE_TOKEN= + +# ── Durability: @workflow/world-postgres ──────────────────────────────────── +# NOTE: EVE_WORKFLOW_WORLD is read at BUILD time (eve compiles agent.ts), so it +# must be set before `bun run build` / `bun run dev` — a runtime-only value is +# silently ignored and you stay on the on-disk world. The Dockerfile bakes it in +# for production, so you do NOT need to set it on Railway. +# Leave it unset locally to use eve's zero-config on-disk world (no Postgres). +# EVE_WORKFLOW_WORLD=@workflow/world-postgres +# Railway's Postgres plugin provides DATABASE_URL automatically. +# DATABASE_URL=postgres://user:pass@host:5432/db # or WORKFLOW_POSTGRES_URL +# Public base URL of THIS service, so durable-run callbacks reach +# /.well-known/workflow/v1/flow. On Railway: https://.up.railway.app +# WORKFLOW_LOCAL_BASE_URL= +# Tuning (silences the graphile-worker pool=24. Bun is used only as the +# package manager (fast installs + bun.lock), matching the rest of the repo. +# Keeping Node here means local and container run the same runtime. +# +# The Railway service's root directory is apps/slack-agent, so the build +# context is this folder (no monorepo involvement). +FROM node:24-slim + +# Bun, for `bun install` only. +COPY --from=oven/bun:1.3-slim /usr/local/bin/bun /usr/local/bin/bun + +WORKDIR /app + +# Install deps first for layer caching, from the committed bun.lock. +COPY package.json bun.lock ./ +RUN bun install --frozen-lockfile + +COPY tsconfig.json ./ +COPY agent ./agent + +# IMPORTANT: eve resolves `experimental.workflow.world` at BUILD time (agent.ts is +# compiled into eve's manifest), not at runtime. This must be set before the build +# or the image silently falls back to the ephemeral on-disk local world. +ENV EVE_WORKFLOW_WORLD=@workflow/world-postgres + +# IMPORTANT: the Maple MCP connection URL is built from MAPLE_API_BASE_URL and is +# also baked into eve's manifest at BUILD time. Railway exposes service variables +# as Docker build args when they are declared with ARG, so set MAPLE_API_BASE_URL +# as a service variable and it becomes available here. If unset at build, the URL +# silently bakes to the local-dev default (https://api.localhost). +ARG MAPLE_API_BASE_URL +ENV MAPLE_API_BASE_URL=${MAPLE_API_BASE_URL} + +RUN npm run build + +COPY docker-entrypoint.sh ./ +RUN chmod +x docker-entrypoint.sh + +ENV NODE_ENV=production +# 8080 everywhere: matches the PORT Railway injects at runtime (which would otherwise +# silently override a different value here), and keeps local `docker run` identical to prod. +# If you change this, change it in all four places: here, EXPOSE, docker-entrypoint.sh's +# fallback, and WORKFLOW_LOCAL_BASE_URL / the generated domain's port in README step 2. +ENV PORT=8080 +EXPOSE 8080 + +CMD ["./docker-entrypoint.sh"] diff --git a/apps/slack-agent/README.md b/apps/slack-agent/README.md new file mode 100644 index 00000000..d7805fd9 --- /dev/null +++ b/apps/slack-agent/README.md @@ -0,0 +1,383 @@ +# @maple/slack-agent + +A general-purpose Slack agent built on the [eve](https://eve.dev) framework, **self-deployed to +Railway** (no Vercel). It answers `@mentions` and DMs, runs tools, and keeps durable multi-turn +sessions. + +It is **multi-workspace**: a single Slack app is distributed to many workspaces, and each +workspace is linked to one Maple organization. Per-team installs (bot token + Maple API key) live +in Maple's API; the agent resolves them per request and talks to Maple's MCP server for +observability tools. See [Multi-workspace architecture](#multi-workspace-architecture). + +## Architecture + +| Concern | Choice | Why | +| --- | --- | --- | +| Framework | eve `0.25.x` (durable agent runtime, Nitro HTTP host) | filesystem-first agents | +| Host | **Railway** container running `eve start` (long-running Node) | eve's supported self-host model; edge Workers is blocked today by a workflow-world protocol gap | +| Model | **Cloudflare Workers AI** via REST (`workers-ai-provider`), `@cf/zai-org/glm-5.2` | `createWorkersAI({ accountId, apiKey })` → an AI-SDK model, no Workers runtime needed; streams structured tool calls, 256K window (see Notes) | +| Durability | **`@workflow/world-postgres`** (`5.0.0-beta.27`) + Railway Postgres | protocol-compatible with eve's vendored `@workflow/*` 5.0.0-beta line | +| Slack | **self-managed, multi-workspace** (`slackChannel()` + custom `webhookVerifier` + per-team `botToken`) | one public app across many workspaces; static signing secret verifies inbound, per-team bot token resolved from the Maple API — see [Multi-workspace architecture](#multi-workspace-architecture) | +| Maple | **resolve endpoint** (`/internal/slack/workspaces/:teamId`) + **MCP** (`/mcp`) | per-team install lookup (TTL-cached) and observability tools scoped per org | + +Key routes (all served by the one container): `POST /eve/v1/session`, `GET /eve/v1/session/:id/stream`, +`POST /eve/v1/slack` (Slack webhook), `GET /eve/v1/health`, and workflow callbacks under +`/.well-known/workflow/v1/flow`. + +## Project layout + +``` +agent/ + agent.ts # model (Workers AI) + workflow world selection + instructions.md # system prompt (Maple domain guidance) + lib/maple.ts # Maple resolve client (TTL cache), Slack sig verify, team→token bridge + channels/slack.ts # multi-workspace Slack channel (webhookVerifier + per-team botToken) + channels/eve.ts # auth policy for the browser/API routes + connections/maple.ts # Maple MCP connection (per-workspace API key auth) + tools/get_time.ts # sample tool proving the tool loop +Dockerfile # node:24-slim (+bun for installs), eve build, entrypoint +docker-entrypoint.sh # runs the Postgres-world migration, then `eve start` +railway.json # DOCKERFILE builder, /eve/v1/health healthcheck +``` + +> **Monorepo note:** this app uses **bun** (like the rest of the repo) but is deliberately a +> **standalone bun project, excluded from the bun workspace** (`"!apps/slack-agent"` in the root +> `package.json`). That keeps `eve dev`'s interactive TUI out of `bun dev`/`turbo`, and keeps the +> Docker build hermetic (context = this folder only). It has its own `bun.lock`. + +> **Runtime: Node, package manager: bun.** eve runs on **Node ≥24** (it hard-fails below that), and +> **cannot run on bun** — `eve dev`'s HMR server uses `crossws`' Node adapter, which throws +> `[crossws] Using Node.js adapter in an incompatible environment` under bun. Production `.output` +> *does* happen to run on bun, but we deliberately use Node in both places so local and container +> match. Bun is still the package manager (`bun install`, `bun.lock`), and `bun run