Clean cloud service boundaries

Rhys Sullivan committed 6 months ago

Commit 586955b1 parent b039b6e3 files · +116 −55Browse files

Unified · Split

3 files

apps/cloud/src/mcp-flow.test.ts +20 −6
apps/cloud/src/services/autumn.ts +19 −13
apps/cloud/src/services/slack.ts +77 −36

apps/cloud/src/mcp-flow.test.ts+20−6

@@ -137,7 +137,7 @@ const seedOrg = async (id: string, name = "MCP Flow Org"): Promise<void> => { · 136 unmodified lines
137	137	beforeAll(() => {
138	138	// Env presence guard — avoids confusing errors downstream if the test
139	139	// wrangler forgot to bind something the DO needs.
140		− if (!env.MCP_SESSION) throw new Error("MCP_SESSION binding missing from test wrangler");
	140	+ expect(env.MCP_SESSION, "MCP_SESSION binding missing from test wrangler").toBeDefined();
141	141	});
142	142
143	143	afterAll(() => undefined);
@@ -384,18 +384,32 @@ describe("/mcp session restore", () => { · 240 unmodified lines
384	384	const getResponse = await mcpGet({ bearer, sessionId: sessionId! });
385	385	expect(getResponse.status).toBe(200);
386	386	expect(getResponse.headers.get("content-type") ?? "").toContain("text/event-stream");
387		− await getResponse.body?.cancel().catch(() => undefined);
	387	+ const responseBody = getResponse.body;
	388	+ if (responseBody) {
	389	+ await Effect.runPromise(
	390	+ Effect.ignore(
	391	+ Effect.tryPromise({
	392	+ try: () => responseBody.cancel(),
	393	+ catch: () => "ResponseBodyCancelFailed" as const,
	394	+ }),
	395	+ ),
	396	+ );
	397	+ }
388	398
389		− const response = await Promise.race([
	399	+ const postResult = await Promise.race([
390	400	mcpPost({
391	401	bearer,
392	402	sessionId,
393	403	body: TOOLS_LIST_REQUEST,
394		− }),
395		− new Promise<never>((_, reject) =>
396		− setTimeout(() => reject(new Error("POST did not return after GET restore")), 5_000),
	404	+ }).then((response) => ({ kind: "response" as const, response })),
	405	+ new Promise<{ readonly kind: "timeout" }>((resolve) =>
	406	+ setTimeout(() => resolve({ kind: "timeout" }), 5_000),
397	407	),
398	408	]);
	409	+ expect(postResult).toEqual(expect.objectContaining({ kind: "response" }));
	410	+ if (postResult.kind !== "response") return;
	411	+
	412	+ const response = postResult.response;
399	413	expect(response.status).toBe(200);
400	414	expect(response.headers.get("content-type") ?? "").toContain("application/json");
401	415	const body = (await response.json()) as {

apps/cloud/src/services/autumn.ts+19−13

@@ -12,7 +12,8 @@ import { Context, Data, Effect, Layer } from "effect"; · 11 unmodified lines
12	12	// ---------------------------------------------------------------------------
13	13
14	14	export class AutumnError extends Data.TaggedError("AutumnError")<{
15		− cause: unknown;
	15	+ message: string;
	16	+ cause?: unknown;
16	17	}> {}
17	18
18	19	// ---------------------------------------------------------------------------
@@ -38,8 +39,8 @@ const make = Effect.sync(() => { · 19 unmodified lines
38	39	const secretKey = env.AUTUMN_SECRET_KEY;
39	40
40	41	if (!secretKey) {
41		− const notConfigured = Effect.die(
42		− new Error("Autumn not configured — AUTUMN_SECRET_KEY is empty"),
	42	+ const notConfigured = Effect.fail(
	43	+ new AutumnError({ message: "Autumn not configured: AUTUMN_SECRET_KEY is empty" }),
43	44	);
44	45	return {
45	46	use: () => notConfigured,
@@ -52,22 +53,27 @@ const make = Effect.sync(() => { · 6 unmodified lines
52	53	const use = <A>(fn: (client: Autumn) => Promise<A>) =>
53	54	Effect.tryPromise({
54	55	try: () => fn(client),
55		− catch: (cause) => new AutumnError({ cause }),
	56	+ catch: (cause) => new AutumnError({ message: "Autumn SDK request failed", cause }),
56	57	}).pipe(Effect.withSpan(`autumn.${fn.name ?? "use"}`));
57	58
58	59	const trackExecution = (organizationId: string) =>
59	60	Effect.gen(function* () {
60	61	yield* Effect.annotateCurrentSpan({ "autumn.customer.id": organizationId });
61		− const outcome = yield* Effect.result(
62		− use((c) => c.track({ customerId: organizationId, featureId: "executions", value: 1 })),
	62	+ yield* use((c) =>
	63	+ c.track({ customerId: organizationId, featureId: "executions", value: 1 }),
	64	+ ).pipe(
	65	+ Effect.catchTag("AutumnError", (error) =>
	66	+ Effect.gen(function* () {
	67	+ // Silent billing data loss is worth paging on — autumn.trackExecution
	68	+ // is fire-and-forget so the caller doesn't handle it themselves.
	69	+ yield* Effect.sync(() => {
	70	+ console.error("[billing] track failed:", error);
	71	+ Sentry.captureException(error);
	72	+ });
	73	+ yield* Effect.annotateCurrentSpan({ "autumn.track.failed": true });
	74	+ }),
	75	+ ),
63	76	);
64		− if (outcome._tag === "Failure") {
65		− // Silent billing data loss is worth paging on — autumn.trackExecution
66		− // is fire-and-forget so the caller doesn't handle it themselves.
67		− console.error("[billing] track failed:", outcome.failure);
68		− Sentry.captureException(outcome.failure);
69		− yield* Effect.annotateCurrentSpan({ "autumn.track.failed": true });
70		− }
71	77	}).pipe(Effect.withSpan("autumn.trackExecution"));
72	78
73	79	return { use, trackExecution } satisfies IAutumnService;

apps/cloud/src/services/slack.ts+77−36

@@ -5,7 +5,7 @@ · 4 unmodified lines
5	5	// ---------------------------------------------------------------------------
6	6
7	7	import { env } from "cloudflare:workers";
8		−import { Context, Data, Effect, Layer } from "effect";
	8	+import { Context, Data, Effect, Layer, Schema } from "effect";
9	9
10	10	export class SlackError extends Data.TaggedError("SlackError")<{
11	11	method: string;
@@ -40,41 +40,82 @@ const randomSuffix = (): string => { · 28 unmodified lines
40	40	return Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join("");
41	41	};
42	42
43		−type SlackResponse = { ok: boolean; error?: string } & Record<string, unknown>;
	43	+const SlackErrorResponse = Schema.Struct({
	44	+ ok: Schema.Literal(false),
	45	+ error: Schema.optional(Schema.String),
	46	+});
	47	+
	48	+const SlackChannel = Schema.Struct({ id: Schema.String, name: Schema.String });
	49	+
	50	+const SlackPostMessageResponse = Schema.Struct({ ok: Schema.Literal(true) });
	51	+
	52	+const SlackCreateChannelResponse = Schema.Struct({
	53	+ ok: Schema.Literal(true),
	54	+ channel: SlackChannel,
	55	+});
	56	+
	57	+const SlackSharedInviteResponse = Schema.Struct({
	58	+ ok: Schema.Literal(true),
	59	+ invite_id: Schema.String,
	60	+ url: Schema.String,
	61	+});
44	62
45	63	const make = Effect.sync(() => {
46	64	const token = env.SLACK_BOT_TOKEN;
47	65
48	66	if (!token) {
49	67	const notConfigured = (method: string) =>
50		− Effect.fail(
51		− new SlackError({ method, error: "SLACK_BOT_TOKEN is not configured" }),
52		− );
	68	+ Effect.fail(new SlackError({ method, error: "SLACK_BOT_TOKEN is not configured" }));
53	69	return {
54	70	createConnectInvite: () => notConfigured("createConnectInvite"),
55	71	} satisfies ISlackService;
56	72	}
57	73
58		− const call = <A extends SlackResponse>(method: string, body: Record<string, unknown>) =>
59		− Effect.tryPromise({
60		− try: async (): Promise<A> => {
61		− const res = await fetch(`https://slack.com/api/${method}`, {
62		− method: "POST",
63		− headers: {
64		− "content-type": "application/json; charset=utf-8",
65		− authorization: `Bearer ${token}`,
66		− },
67		− body: JSON.stringify(body),
68		− });
69		− const json = (await res.json()) as A;
70		− if (!json.ok) throw new Error(json.error ?? "unknown_slack_error");
71		− return json;
72		− },
73		− catch: (cause) =>
74		− new SlackError({
	74	+ const call = <A extends { ok: true }>(
	75	+ method: string,
	76	+ body: Record<string, unknown>,
	77	+ successSchema: Schema.Decoder<A>,
	78	+ ) =>
	79	+ Effect.gen(function* () {
	80	+ const json = yield* Effect.tryPromise({
	81	+ try: async (): Promise<unknown> => {
	82	+ const res = await fetch(`https://slack.com/api/${method}`, {
	83	+ method: "POST",
	84	+ headers: {
	85	+ "content-type": "application/json; charset=utf-8",
	86	+ authorization: `Bearer ${token}`,
	87	+ },
	88	+ body: JSON.stringify(body),
	89	+ });
	90	+ return res.json();
	91	+ },
	92	+ catch: () =>
	93	+ new SlackError({
	94	+ method,
	95	+ error: "Failed to read Slack API response",
	96	+ }),
	97	+ });
	98	+
	99	+ const response = yield* Schema.decodeUnknownEffect(
	100	+ Schema.Union([successSchema, SlackErrorResponse]),
	101	+ )(json).pipe(
	102	+ Effect.mapError(
	103	+ () =>
	104	+ new SlackError({
	105	+ method,
	106	+ error: "Unexpected Slack API response",
	107	+ }),
	108	+ ),
	109	+ );
	110	+
	111	+ if (!response.ok) {
	112	+ return yield* new SlackError({
75	113	method,
76		− error: cause instanceof Error ? cause.message : String(cause),
77		− }),
	114	+ error: response.error ?? "unknown_slack_error",
	115	+ });
	116	+ }
	117	+
	118	+ return response;
78	119	}).pipe(Effect.withSpan(`slack.${method}`));
79	120
80	121	const createConnectInvite: ISlackService["createConnectInvite"] = ({
@@ -87,10 +128,7 @@ const make = Effect.sync(() => { · 6 unmodified lines
87	128	// Slack channel names: lowercase, no spaces, max 80 chars, unique per workspace.
88	129	const baseName = `shared-${slugifyEmail(email)}`.slice(0, 80);
89	130	const tryCreate = (n: string) =>
90		− call<SlackResponse & { channel: { id: string; name: string } }>(
91		− "conversations.create",
92		− { name: n, is_private: false },
93		− );
	131	+ call("conversations.create", { name: n, is_private: false }, SlackCreateChannelResponse);
94	132
95	133	const created = yield* tryCreate(baseName).pipe(
96	134	Effect.catchTag("SlackError", (err) =>
@@ -108,12 +146,16 @@ const make = Effect.sync(() => { · 11 unmodified lines
108	146	note ? `Note: ${note}` : null,
109	147	].filter(Boolean) as string[];
110	148
111		− yield* call<SlackResponse>("chat.postMessage", {
112		− channel: channel.id,
113		− text: helloLines.join("\n"),
114		− });
	149	+ yield* call(
	150	+ "chat.postMessage",
	151	+ {
	152	+ channel: channel.id,
	153	+ text: helloLines.join("\n"),
	154	+ },
	155	+ SlackPostMessageResponse,
	156	+ );
115	157
116		− const invite = yield* call<SlackResponse & { invite_id: string; url: string }>(
	158	+ const invite = yield* call(
117	159	"conversations.inviteShared",
118	160	{
119	161	channel: channel.id,
@@ -122,15 +164,14 @@ const make = Effect.sync(() => { · 2 unmodified lines
122	164	// which is what we want for a focused 1:1 support conversation.
123	165	external_limited: true,
124	166	},
	167	+ SlackSharedInviteResponse,
125	168	);
126	169
127	170	return {
128	171	channel: { id: channel.id, name: channel.name },
129	172	invite: { invite_id: invite.invite_id, url: invite.url },
130	173	};
131		− }).pipe(
132		− Effect.withSpan("slack.createConnectInvite", { attributes: { "slack.email": email } }),
133		− );
	174	+ }).pipe(Effect.withSpan("slack.createConnectInvite", { attributes: { "slack.email": email } }));
134	175
135	176	return { createConnectInvite } satisfies ISlackService;
136	177	});

Clean cloud service boundaries

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code