Skip to content

Commit 91e89dc

Browse files
dengzhaofundengzhaofun
authored
feat(analytics): server 全项目 Tinybird 埋点覆盖 (#45)
1 parent c10d4dc commit 91e89dc

39 files changed

Lines changed: 1774 additions & 88 deletions

‎apps/server/src/index.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { secureHeaders } from "hono/secure-headers";
99
import { auth } from "./auth";
1010
import { endUserAuth, EU_ORG_ID_HEADER } from "./end-user-auth";
1111
import type { HonoEnv } from "./env";
12+
import { requestContext } from "./lib/request-context";
1213
import { requireClientCredential } from "./middleware/require-client-credential";
1314
import { requestLog } from "./middleware/request-log";
1415
import { session } from "./middleware/session";
@@ -171,6 +172,13 @@ app.on(["POST", "GET"], "/api/client/auth/*", (c) => {
171172

172173
// Inject c.var.user / c.var.session for downstream business routes
173174
app.use("*", session);
175+
// Put the per-request AsyncLocalStorage store in place so domain-event
176+
// subscribers can stamp Tinybird rows with the same `traceId` that
177+
// `http_requests` records. Must wrap everything AFTER `requestId()` has
178+
// run but BEFORE any handler may emit events.
179+
app.use("*", (c, next) =>
180+
requestContext.run({ traceId: c.get("requestId") }, next),
181+
);
174182
// Auto-ingest every request into Tinybird's http_requests dataset.
175183
// Must run AFTER session so we know which tenant to tag.
176184
app.use("*", requestLog);

‎apps/server/src/lib/request-context.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/**
2+
* Per-request AsyncLocalStorage (ALS) store.
3+
*
4+
* Why this exists: services are protocol-agnostic (see apps/server/CLAUDE.md
5+
* — service.ts must not import Hono or Context). But downstream subscribers
6+
* like `modules/analytics/subscribers/*` need the per-request `traceId` to
7+
* tag every business event in Tinybird so they join with the `http_requests`
8+
* row recorded by `middleware/request-log.ts`. Threading `traceId` through
9+
* every service signature and every `emit()` payload would bloat 26 existing
10+
* domain events for a single cross-cutting concern — ALS keeps that concern
11+
* invisible to the business layer.
12+
*
13+
* Workers runtime supports `node:async_hooks` under `nodejs_compat`
14+
* (enabled in `wrangler.jsonc`). The store is populated by a middleware
15+
* in `src/index.ts` after `requestId()` has assigned `c.get("requestId")`,
16+
* and by `src/scheduled.ts` on each cron tick.
17+
*
18+
* Read with `getTraceId()` — returns `""` when outside a store (e.g. under
19+
* vitest where nothing populated it). This matches the wire-format default
20+
* the Tinybird writer uses for empty trace ids.
21+
*/
22+
23+
import { AsyncLocalStorage } from "node:async_hooks";
24+
25+
export type RequestContext = {
26+
traceId: string;
27+
};
28+
29+
export const requestContext = new AsyncLocalStorage<RequestContext>();
30+
31+
export function getTraceId(): string {
32+
return requestContext.getStore()?.traceId ?? "";
33+
}

‎apps/server/src/modules/activity/service.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,16 @@ declare module "../../lib/event-bus" {
131131
endUserId: string;
132132
milestoneAlias: string;
133133
};
134+
"activity.joined": {
135+
organizationId: string;
136+
activityId: string;
137+
activityAlias: string | null;
138+
endUserId: string;
139+
// True when this call was the first-ever join (upsert inserted);
140+
// false when the user was re-marking `lastActiveAt`. Downstream
141+
// analytics can filter to first-time participation.
142+
firstTime: boolean;
143+
};
134144
}
135145
}
136146

@@ -650,6 +660,23 @@ export function createActivityService(
650660
set: { lastActiveAt: now },
651661
})
652662
.returning();
663+
664+
// Insert vs update discrimination: the upsert's `set` only touches
665+
// `lastActiveAt`, so an existing row retains its original
666+
// `joinedAt`. A fresh insert stamps `joinedAt` to `now`, so
667+
// equality against `now` reliably marks the first-time join.
668+
const firstTime = row!.joinedAt.getTime() === now.getTime();
669+
670+
if (events) {
671+
await events.emit("activity.joined", {
672+
organizationId: params.organizationId,
673+
activityId: activity.id,
674+
activityAlias: activity.alias,
675+
endUserId: params.endUserId,
676+
firstTime,
677+
});
678+
}
679+
653680
return row!;
654681
},
655682

‎apps/server/src/modules/analytics/subscribers.ts‎

Lines changed: 0 additions & 43 deletions
This file was deleted.

‎apps/server/src/modules/analytics/subscribers/activity.ts‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
import type { AnalyticsService } from "../../../lib/analytics";
2+
import type { EventBus } from "../../../lib/event-bus";
3+
4+
import { makeWriteEvent } from "./utils";
5+
6+
export function registerActivitySubscribers(
7+
events: EventBus,
8+
analytics: AnalyticsService,
9+
): void {
10+
const write = makeWriteEvent(analytics);
11+
12+
events.on("activity.state.changed", (p) => {
13+
write({
14+
orgId: p.organizationId,
15+
event: "activity.state.changed",
16+
source: "activity",
17+
eventData: {
18+
activityId: p.activityId,
19+
previousState: p.previousState,
20+
newState: p.newState,
21+
},
22+
});
23+
});
24+
25+
events.on("activity.schedule.fired", (p) => {
26+
write({
27+
orgId: p.organizationId,
28+
event: "activity.schedule.fired",
29+
source: "activity",
30+
eventData: {
31+
activityId: p.activityId,
32+
scheduleAlias: p.scheduleAlias,
33+
actionType: p.actionType,
34+
firedAt: p.firedAt,
35+
actionConfig: p.actionConfig,
36+
},
37+
});
38+
});
39+
40+
events.on("activity.milestone.claimed", (p) => {
41+
write({
42+
orgId: p.organizationId,
43+
endUserId: p.endUserId,
44+
event: "activity.milestone.claimed",
45+
source: "activity",
46+
amount: 1,
47+
eventData: {
48+
activityId: p.activityId,
49+
milestoneAlias: p.milestoneAlias,
50+
},
51+
});
52+
});
53+
54+
events.on("activity.joined", (p) => {
55+
write({
56+
orgId: p.organizationId,
57+
endUserId: p.endUserId,
58+
event: "activity.joined",
59+
source: "activity",
60+
// amount = 1 if this is a first-time join, 0 if it was just a
61+
// lastActiveAt refresh. Sum(amount) gives unique participants;
62+
// count(*) gives all touches.
63+
amount: p.firstTime ? 1 : 0,
64+
eventData: {
65+
activityId: p.activityId,
66+
activityAlias: p.activityAlias,
67+
firstTime: p.firstTime,
68+
},
69+
});
70+
});
71+
}

‎apps/server/src/modules/analytics/subscribers/announcement.ts‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
import type { AnalyticsService } from "../../../lib/analytics";
2+
import type { EventBus } from "../../../lib/event-bus";
3+
4+
import { makeWriteEvent } from "./utils";
5+
6+
export function registerAnnouncementSubscribers(
7+
events: EventBus,
8+
analytics: AnalyticsService,
9+
): void {
10+
const write = makeWriteEvent(analytics);
11+
12+
events.on("announcement.created", (p) => {
13+
write({
14+
orgId: p.organizationId,
15+
event: "announcement.created",
16+
source: "announcement",
17+
eventData: {
18+
announcementId: p.announcementId,
19+
alias: p.alias,
20+
kind: p.kind,
21+
},
22+
});
23+
});
24+
25+
events.on("announcement.updated", (p) => {
26+
write({
27+
orgId: p.organizationId,
28+
event: "announcement.updated",
29+
source: "announcement",
30+
eventData: {
31+
announcementId: p.announcementId,
32+
alias: p.alias,
33+
},
34+
});
35+
});
36+
37+
events.on("announcement.deleted", (p) => {
38+
write({
39+
orgId: p.organizationId,
40+
event: "announcement.deleted",
41+
source: "announcement",
42+
eventData: {
43+
announcementId: p.announcementId,
44+
alias: p.alias,
45+
},
46+
});
47+
});
48+
49+
events.on("announcement.impression", (p) => {
50+
write({
51+
orgId: p.organizationId,
52+
endUserId: p.endUserId,
53+
event: "announcement.impression",
54+
source: "announcement",
55+
amount: 1,
56+
eventData: {
57+
announcementId: p.announcementId,
58+
alias: p.alias,
59+
kind: p.kind,
60+
},
61+
});
62+
});
63+
64+
events.on("announcement.click", (p) => {
65+
write({
66+
orgId: p.organizationId,
67+
endUserId: p.endUserId,
68+
event: "announcement.click",
69+
source: "announcement",
70+
amount: 1,
71+
eventData: {
72+
announcementId: p.announcementId,
73+
alias: p.alias,
74+
ctaUrl: p.ctaUrl,
75+
},
76+
});
77+
});
78+
}

‎apps/server/src/modules/analytics/subscribers/assist-pool.ts‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
import type { AnalyticsService } from "../../../lib/analytics";
2+
import type { EventBus } from "../../../lib/event-bus";
3+
4+
import { makeWriteEvent } from "./utils";
5+
6+
export function registerAssistPoolSubscribers(
7+
events: EventBus,
8+
analytics: AnalyticsService,
9+
): void {
10+
const write = makeWriteEvent(analytics);
11+
12+
events.on("assist_pool.instance_created", (p) => {
13+
write({
14+
orgId: p.organizationId,
15+
// `endUserId` is the pool owner — the one who "initiated" it.
16+
endUserId: p.endUserId,
17+
event: "assist_pool.instance_created",
18+
source: "assist-pool",
19+
amount: p.targetAmount,
20+
eventData: {
21+
configId: p.configId,
22+
instanceId: p.instanceId,
23+
expiresAt: p.expiresAt,
24+
},
25+
});
26+
});
27+
28+
events.on("assist_pool.contributed", (p) => {
29+
write({
30+
orgId: p.organizationId,
31+
// Session actor = the contributor (initiator), not the pool owner.
32+
// Keeps per-user contribution timeseries queryable.
33+
endUserId: p.initiatorEndUserId,
34+
event: "assist_pool.contributed",
35+
source: "assist-pool",
36+
amount: p.amount,
37+
eventData: {
38+
configId: p.configId,
39+
instanceId: p.instanceId,
40+
poolOwnerEndUserId: p.endUserId,
41+
remaining: p.remaining,
42+
},
43+
});
44+
});
45+
46+
events.on("assist_pool.completed", (p) => {
47+
write({
48+
orgId: p.organizationId,
49+
endUserId: p.endUserId,
50+
event: "assist_pool.completed",
51+
source: "assist-pool",
52+
eventData: {
53+
configId: p.configId,
54+
instanceId: p.instanceId,
55+
rewards: p.rewards,
56+
},
57+
});
58+
});
59+
60+
events.on("assist_pool.expired", (p) => {
61+
write({
62+
orgId: p.organizationId,
63+
endUserId: p.endUserId,
64+
event: "assist_pool.expired",
65+
// `outcome` stays "ok" — expiry is a normal lifecycle terminal
66+
// state, not a system error. The reason is queryable via
67+
// `event_data.reason` ("timeout" | "force").
68+
source: "assist-pool",
69+
eventData: {
70+
configId: p.configId,
71+
instanceId: p.instanceId,
72+
reason: p.reason,
73+
},
74+
});
75+
});
76+
}

0 commit comments

Comments
 (0)