103 lines
4.4 KiB
TypeScript
103 lines
4.4 KiB
TypeScript
import { config } from "../config.js";
|
|
import { log } from "../log.js";
|
|
import { recordGrowEvent } from "./record-grow-event.js";
|
|
import { routeGrowEventToUserActor } from "./route-to-user-actor.js";
|
|
|
|
type RedisMessage = { id: string; message: Record<string, string> };
|
|
type RedisStreamResponse = Array<{ name: string; messages: RedisMessage[] }>;
|
|
|
|
type RedisClientLike = {
|
|
connect: () => Promise<void>;
|
|
on: (event: string, listener: (err: unknown) => void) => void;
|
|
xGroupCreate: (stream: string, group: string, id: string, opts?: Record<string, unknown>) => Promise<unknown>;
|
|
xReadGroup: (group: string, consumer: string, streams: { key: string; id: string }, opts?: Record<string, unknown>) => Promise<RedisStreamResponse | null>;
|
|
xAck: (stream: string, group: string, id: string) => Promise<unknown>;
|
|
};
|
|
|
|
async function loadRedisCreateClient(): Promise<(opts: { url: string }) => RedisClientLike> {
|
|
const dynamicImport = Function("specifier", "return import(specifier)") as (specifier: string) => Promise<unknown>;
|
|
const mod = (await dynamicImport("redis")) as { createClient?: (opts: { url: string }) => RedisClientLike };
|
|
if (!mod.createClient) throw new Error("redis package did not expose createClient");
|
|
return mod.createClient;
|
|
}
|
|
|
|
function parseFieldValue(value: string): unknown {
|
|
if (!value) return value;
|
|
if ((value.startsWith("{") && value.endsWith("}")) || (value.startsWith("[") && value.endsWith("]"))) {
|
|
try { return JSON.parse(value); } catch { return value; }
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function fieldsToEvent(fields: Record<string, string>, stream: string) {
|
|
const payload = fields.payload ? parseFieldValue(fields.payload) : parseFieldValue(fields.data ?? "{}");
|
|
const correlation = fields.correlation ? parseFieldValue(fields.correlation) : undefined;
|
|
const subject = fields.subject ? parseFieldValue(fields.subject) : undefined;
|
|
const mission = fields.mission ? parseFieldValue(fields.mission) : undefined;
|
|
return {
|
|
id: fields.id,
|
|
source: fields.source ?? stream,
|
|
type: fields.type ?? fields.event_type ?? "service.event",
|
|
category: fields.category ?? "service",
|
|
userId: fields.userId ?? fields.user_id,
|
|
orgId: fields.orgId ?? fields.org_id,
|
|
occurredAt: fields.occurredAt ?? fields.occurred_at ?? fields.timestamp,
|
|
mission,
|
|
subject,
|
|
correlation,
|
|
payload: payload && typeof payload === "object" && !Array.isArray(payload) ? payload : { value: payload },
|
|
raw: fields,
|
|
dedupeKey: fields.dedupeKey ?? fields.dedupe_key,
|
|
};
|
|
}
|
|
|
|
export async function startGrowEventsRedisConsumer() {
|
|
if (!config.growEventsRedisUrl) {
|
|
log.info("grow events Redis consumer disabled (GROW_EVENTS_REDIS_URL/REDIS_URL not set)");
|
|
return;
|
|
}
|
|
|
|
const createClient = await loadRedisCreateClient();
|
|
const redis = createClient({ url: config.growEventsRedisUrl });
|
|
redis.on("error", (err) => log.warn({ err }, "grow events redis client error"));
|
|
await redis.connect();
|
|
|
|
try {
|
|
await redis.xGroupCreate(config.growEventsStream, config.growEventsConsumerGroup, "0", { MKSTREAM: true });
|
|
log.info({ stream: config.growEventsStream, group: config.growEventsConsumerGroup }, "created grow events consumer group");
|
|
} catch (err) {
|
|
if (!String(err).includes("BUSYGROUP")) throw err;
|
|
}
|
|
|
|
log.info({ stream: config.growEventsStream, group: config.growEventsConsumerGroup }, "grow events redis consumer started");
|
|
|
|
void (async () => {
|
|
while (true) {
|
|
try {
|
|
const response = await redis.xReadGroup(
|
|
config.growEventsConsumerGroup,
|
|
config.growEventsConsumerName,
|
|
{ key: config.growEventsStream, id: ">" },
|
|
{ COUNT: 50, BLOCK: 5000 },
|
|
);
|
|
if (!response) continue;
|
|
for (const stream of response) {
|
|
for (const message of stream.messages) {
|
|
try {
|
|
const event = await recordGrowEvent(fieldsToEvent(message.message, stream.name));
|
|
await routeGrowEventToUserActor(event);
|
|
} catch (err) {
|
|
log.error({ err, messageId: message.id }, "failed to ingest grow event stream message");
|
|
} finally {
|
|
await redis.xAck(stream.name, config.growEventsConsumerGroup, message.id);
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
log.error({ err }, "grow events redis consumer loop error");
|
|
await new Promise((resolve) => setTimeout(resolve, 2000));
|
|
}
|
|
}
|
|
})();
|
|
}
|