diff --git a/webapps/console/components/DataView/EventsBrowser.tsx b/webapps/console/components/DataView/EventsBrowser.tsx index 6b20b83fe..9f1ea132d 100644 --- a/webapps/console/components/DataView/EventsBrowser.tsx +++ b/webapps/console/components/DataView/EventsBrowser.tsx @@ -72,6 +72,13 @@ type EventsBrowserState = { error?: string; }; +const ALL_ACTORS = "all"; +/** + * Pseudo-entity for the "no connection selected" case: events of every connection of the workspace. + * `mode` is the default for the bulker view where the mode selector is bound to the selected entity + */ +const allConnectionsEntity = { id: ALL_ACTORS, name: "All Connections", type: "all", mode: "batch" }; + const defaultState: EventsBrowserState = { bulkerMode: undefined, eventsLoading: false, @@ -213,6 +220,7 @@ const EventsBrowser0 = ({ ? streams : streamType == "function" ? [ + allConnectionsEntity, ...mappedConnections, ...profileBuilders .filter(p => p.version > 0) @@ -225,7 +233,7 @@ const EventsBrowser0 = ({ ] : streamType == "dead-letter" ? [ - { id: "all", name: "All Connections", type: "all" }, + allConnectionsEntity, ...mappedConnections, ...profileBuilders .filter(p => p.version > 0) @@ -237,6 +245,7 @@ const EventsBrowser0 = ({ }), ] : [ + allConnectionsEntity, ...mappedConnections.filter(link => link.usesBulker || link.hybrid), ...(hasActiveProfileBuilder ? destinations @@ -358,7 +367,10 @@ const EventsBrowser0 = ({ setConnection(connection); setDebugEnabled(new Date(connection.data.debugTill) > new Date()); } else { + //"All Connections" or a profile builder - debug belongs to a single connection, and the + //banner would otherwise keep announcing it for a connection that is no longer selected setConnection(undefined); + setDebugEnabled(false); } })(); } @@ -420,7 +432,8 @@ const EventsBrowser0 = ({ const data = await eventsLogApi.get( `${eventsLogStream}`, level === "all" ? "all" : [level], - actorId, + //no actor selected - events of all the connections of the workspace + actorId === ALL_ACTORS ? undefined : actorId, { start: dates && dates[0] ? new Date(dates[0]) : undefined, end: beforeDate || (dates && dates[1] ? new Date(dates[1]) : undefined), @@ -735,6 +748,7 @@ const EventsBrowser0 = ({ entityType={entityType} actorId={actorId} mappedConnections={mappedConnectionsMap} + actorsMap={entitiesMap} events={shownEvents} loadEvents={() => dispatch({ @@ -762,10 +776,59 @@ type TableProps = { entityType: string; actorId: string; mappedConnections: Record; + /** All entities that can be selected in the connection selector, including profile builders */ + actorsMap?: Record; loadEvents: () => void; }; -const FunctionsLogTable = ({ loadEvents, loading, streamType, entityType, actorId, events }: TableProps) => { +/** Both sides of a connection must fit into a narrow column - see TruncationPolicy */ +const actorTruncation = 13; + +/** + * Renders a link to the connection (or profile builder) an event belongs to. Used for the extra + * column that is shown when events of all connections are displayed at once + */ +const ActorLink: React.FC<{ actorId?: string; actorsMap?: Record }> = ({ actorId, actorsMap }) => { + if (!actorId) { + return null; + } + const actor = actorsMap?.[actorId]; + if (!actor || actor.type === "all") { + return {trimMiddle(actorId, 16)}; + } + const title = + actor.type === "profile-builder" ? ( + + ) : ( + + ); + const href = actor.type === "profile-builder" ? `/profile-builder` : `/connections/edit?id=${actorId}`; + return ( + + {/* stopPropagation - otherwise a click opens the event drawer along with the navigation */} + e.stopPropagation()}> + {title} + + + ); +}; + +/** Column shown instead of the connection selector when events of all connections are displayed */ +const actorColumn = (actorsMap?: Record) => ({ + title: "Connection", + width: "20em", + dataIndex: "actorId", + key: "actorId", + render: (actorId: string) => , +}); + +const FunctionsLogTable = ({ loadEvents, loading, streamType, entityType, actorId, events, actorsMap }: TableProps) => { const workspace = useWorkspace(); const [funcsMap, setFuncsMap] = useState>({}); @@ -812,6 +875,7 @@ const FunctionsLogTable = ({ loadEvents, loading, streamType, entityType, actorI width: "13em", render: d => , }, + ...(actorId === ALL_ACTORS ? [actorColumn(actorsMap) as any] : []), { title: "Function", width: "14em", @@ -916,7 +980,7 @@ const FunctionsLogTable = ({ loadEvents, loading, streamType, entityType, actorI ); }; -const StreamEventsTable = ({ loadEvents, loading, streamType, entityType, actorId, events }: TableProps) => { +const StreamEventsTable = ({ loadEvents, loading, streamType, entityType, actorId, events, actorsMap }: TableProps) => { const streamEvents = events ? events.map((e, i) => { e = { @@ -939,6 +1003,7 @@ const StreamEventsTable = ({ loadEvents, loading, streamType, entityType, actorI width: "13em", render: d => , }, + ...(actorId === ALL_ACTORS ? [actorColumn(actorsMap) as any] : []), { title: "Queue size", width: "7em", @@ -1012,7 +1077,36 @@ const StreamEventsTable = ({ loadEvents, loading, streamType, entityType, actorI ); }; -const BatchTable = ({ loadEvents, loading, streamType, entityType, actorId, events }: TableProps) => { +const estimatedCostNote = + "Estimation of how much the data warehouse charges for this load. The exact amount may be different due to " + + "discounts, committed use pricing, free tiers, etc."; + +const EstimatedCostHeader: React.FC<{}> = () => ( + + + Est. cost + + +); + +const formatCost = (cost: number) => + cost < 0.0001 + ? "< $0.0001" + : "$" + cost.toLocaleString("en-US", { minimumFractionDigits: 2, maximumFractionDigits: 4 }); + +const EstimatedCost: React.FC<{ cost?: number }> = ({ cost }) => { + //nothing charged (or nothing reported) - an empty cell is less noisy than a row of $0 + if (typeof cost !== "number" || isNaN(cost) || cost === 0) { + return null; + } + return ( + + {formatCost(cost)} + + ); +}; + +const BatchTable = ({ loadEvents, loading, streamType, entityType, actorId, events, actorsMap }: TableProps) => { const batchEvents = (events || ([] as EventsLogRecord[])).map((e, i) => ({ ...e, id: e.date + "_" + i, @@ -1025,6 +1119,7 @@ const BatchTable = ({ loadEvents, loading, streamType, entityType, actorId, even width: "13em", render: d => , }, + ...(actorId === ALL_ACTORS ? [actorColumn(actorsMap) as any] : []), { title: "Batch size", width: "7em", @@ -1037,6 +1132,13 @@ const BatchTable = ({ loadEvents, loading, streamType, entityType, actorId, even dataIndex: ["content", "queueSize"], key: "queue", }, + { + title: , + width: "9em", + dataIndex: ["content", "statistics", "estimatedCost"], + key: "cost", + render: (cost: number | undefined) => , + }, { title: "Status", width: "8em", diff --git a/webapps/console/lib/server/events-log-stream.ts b/webapps/console/lib/server/events-log-stream.ts new file mode 100644 index 000000000..4eb8b95e4 --- /dev/null +++ b/webapps/console/lib/server/events-log-stream.ts @@ -0,0 +1,237 @@ +import type { NextApiResponse } from "next"; +import { z } from "zod"; +import dayjs from "dayjs"; +import utc from "dayjs/plugin/utc"; +import zlib from "zlib"; +import { pipeline } from "node:stream"; +import { db } from "./db"; +import { clickhouse } from "./clickhouse"; +import { getServerLog } from "./log"; +import { ApiError } from "../shared/errors"; +import { SessionUser } from "../schema"; +import { getWorkspace, verifyAccess } from "../api"; + +dayjs.extend(utc); + +const log = getServerLog("events-log"); + +//Vercel Limit: https://vercel.com/docs/functions/streaming-functions#limitations-for-streaming-edge-functions +const maxStreamingResponseSize = 100_000_000; + +/** + * Query params shared by both events-log routes: `log/[type]` (all the actors of the workspace) + * and `log/[type]/[actorId]` (a single actor) + */ +export const eventsLogQuery = z.object({ + type: z.string(), + workspaceId: z.string(), + levels: z.string().optional(), + limit: z.coerce.number().optional().default(50), + start: z.coerce.date().optional(), + end: z.coerce.date().optional(), + //people can search for ISO timestamps. that we automatically convert to date + search: z.any().optional(), +}); + +export type EventsLogQuery = z.infer; + +/** + * Checks that the actor belongs to the workspace. Allowed actor kinds differ by log type: + * `incoming` is emitted by streams, everything else by connections / destinations / profile builders + */ +async function assertActorBelongsToWorkspace(workspaceId: string, actorId: string, type: string) { + if (type === "incoming") { + const source = await db.prisma().configurationObject.findFirst({ where: { id: actorId, workspaceId } }); + if (!source) { + throw new ApiError(`site doesn't belong to the current workspace`, { status: 403 }); + } + return; + } + const [link, pb, dst] = await Promise.all([ + db.prisma().configurationObjectLink.findFirst({ where: { id: actorId, workspaceId } }), + db.prisma().profileBuilder.findFirst({ where: { id: actorId, workspaceId } }), + db.prisma().configurationObject.findFirst({ where: { id: actorId, workspaceId, type: "destination" } }), + ]); + if (!link && !pb && !dst) { + throw new ApiError(`connection doesn't belong to the current workspace`, { status: 403 }); + } +} + +/** + * Which kinds of actor emit each log type, i.e. what an actorId of that type can be: + * - `incoming` → the site the event was sent to (`ingest`, ActorId = stream id) + * - `function` → the connection the function chain ran for, or the profile builder + * (`rotor`, connectionId = link id / `builder.ts` = profile builder id) + * - `bulker_*` → bulker's destinationId: the link id for a regular connection, the destination + * id when the profile builder writes its result + * An unknown type gets every kind - narrower would silently hide rows + */ +function actorKinds(type: string): { + streams: boolean; + links: boolean; + destinations: boolean; + profileBuilders: boolean; +} { + switch (type) { + case "incoming": + return { streams: true, links: false, destinations: false, profileBuilders: false }; + case "function": + return { streams: false, links: true, destinations: false, profileBuilders: true }; + case "bulker_batch": + case "bulker_stream": + return { streams: false, links: true, destinations: true, profileBuilders: false }; + default: + return { streams: true, links: true, destinations: true, profileBuilders: true }; + } +} + +/** + * `events_log` has no workspaceId column — it is scoped by actorId only. To serve the "all actors" + * view we resolve the actor ids the workspace owns and filter by them, otherwise rows of other + * workspaces would leak. Only the kinds that actually emit the requested type are included, so + * unrelated ids (and id collisions across tables) can't widen the result set + */ +async function workspaceActorIds(workspaceId: string, type: string): Promise { + const kinds = actorKinds(type); + const configObjects = async (type: "stream" | "destination") => + ( + await db + .prisma() + .configurationObject.findMany({ where: { workspaceId, type, deleted: false }, select: { id: true } }) + ).map(o => o.id); + + const [streams, links, destinations, pbs] = await Promise.all([ + kinds.streams ? configObjects("stream") : [], + kinds.links + ? db + .prisma() + .configurationObjectLink.findMany({ where: { workspaceId, deleted: false }, select: { id: true } }) + .then(ls => ls.map(l => l.id)) + : [], + kinds.destinations ? configObjects("destination") : [], + kinds.profileBuilders + ? db + .prisma() + .profileBuilder.findMany({ where: { workspaceId }, select: { id: true } }) + .then(ps => ps.map(p => p.id)) + : [], + ]); + return [...new Set([...streams, ...links, ...destinations, ...pbs])]; +} + +/** + * Streams events log records as gzipped ndjson. `actorId` omitted → records of every actor of the + * workspace; each record carries `actorId` so the UI can tell them apart + */ +export async function streamEventsLog({ + user, + res, + query, + actorId, +}: { + user: SessionUser; + res: NextApiResponse; + query: EventsLogQuery; + actorId?: string; +}) { + log.atDebug().log("GET", JSON.stringify({ ...query, actorId }, null, 2)); + await verifyAccess(user, query.workspaceId); + //the route accepts a slug as well as an id (verifyAccess resolves one internally), but every + //lookup below matches on workspaceId - a slug would find no actors and return a false empty + const workspaceId = (await getWorkspace(query.workspaceId)).id; + + let actorIds: string[] = []; + if (actorId) { + await assertActorBelongsToWorkspace(workspaceId, actorId, query.type); + } else { + actorIds = await workspaceActorIds(workspaceId, query.type); + } + + res.writeHead(200, { + "Content-Type": "application/x-ndjson", + "Content-Encoding": "gzip", + }); + + if (!actorId && actorIds.length === 0) { + //nothing to query - the workspace has no actors of this kind. The body still must be a valid + //(empty) gzip stream, since Content-Encoding is already announced + await new Promise(resolve => { + const emptyGzip = zlib.createGzip(); + pipeline(emptyGzip, res, () => resolve()); + emptyGzip.end(); + }); + return; + } + + const sqlQuery = `select timestamp as date, level, actorId, message as content from events_log + where + ${actorId ? "actorId = {actorId:String}" : "actorId in ({actorIds:Array(String)})"} + and type = {type:String} + ${query.levels ? "and level in ({levels:Array(String)})" : ""} + ${query.start ? "and timestamp >= {start:String}" : ""} + ${query.end ? "and timestamp < {end:String}" : ""} + ${query.search ? "and message ilike concat('%',{search:String},'%')" : ""} + order by timestamp desc limit {limit:UInt32}`; + const chResult = await clickhouse.query({ + query: sqlQuery, + query_params: { + actorId, + actorIds: actorId ? undefined : actorIds, + type: query.type, + levels: query.levels ? query.levels.split(",") : undefined, + start: query.start ? dayjs(query.start).utc().format("YYYY-MM-DD HH:mm:ss.SSS") : undefined, + end: query.end ? dayjs(query.end).utc().format("YYYY-MM-DD HH:mm:ss.SSS") : undefined, + search: + typeof query.search === "undefined" + ? undefined + : query.search instanceof Date + ? query.search.toISOString() + : query.search, + limit: query.limit, + }, + format: "JSONEachRow", + clickhouse_settings: { + wait_end_of_query: 1, + }, + }); + var responsePromiseResolve; + let responsePromise = new Promise((resolve, reject) => { + responsePromiseResolve = resolve; + }); + const gzip = zlib.createGzip(); + pipeline(gzip, res, err => { + if (err) { + log.atError().withCause(err).log("Error piping data to response"); + } + responsePromiseResolve(); + }); + const stream = chResult.stream(); + stream.on("data", rs => { + for (const r of rs) { + const row = r.json() as any; + if (gzip.bytesWritten < maxStreamingResponseSize) { + const line = JSON.stringify({ + date: dayjs(row.date).utc(true).toDate(), + level: row.level, + actorId: row.actorId, + content: JSON.parse(row.content), + }); + gzip.write(line + "\n"); + } else { + stream.destroy(); + } + } + }); + stream.on("error", err => { + log.atError().withCause(err).log("Error streaming data"); + gzip.end(); + }); + stream.on("close", () => { + gzip.end(); + }); + stream.on("end", () => { + gzip.end(); + }); + //wait for stream end + await responsePromise; +} diff --git a/webapps/console/lib/server/events-log.ts b/webapps/console/lib/server/events-log.ts index 178cdb98c..8a01dd1e0 100644 --- a/webapps/console/lib/server/events-log.ts +++ b/webapps/console/lib/server/events-log.ts @@ -7,5 +7,7 @@ export type EventsLogFilter = { export type EventsLogRecord = { id: string; date: Date; + /** Set when the log is queried across all the workspace's actors */ + actorId?: string; content: any; }; diff --git a/webapps/console/lib/useApi.ts b/webapps/console/lib/useApi.ts index 975c4421b..165a5d3ca 100644 --- a/webapps/console/lib/useApi.ts +++ b/webapps/console/lib/useApi.ts @@ -23,10 +23,11 @@ export type ConfigApi = { }; export type EventsLogApi = { + /** `actorId` omitted → events of all the actors of the workspace, each record carries `actorId` */ get( eventType: string, levels: ("warn" | "info" | "error" | "debug")[] | "all", - actorId: string, + actorId: string | undefined, filter: EventsLogFilter, limit: number, search?: string @@ -43,13 +44,13 @@ export function getEventsLogApi(workspaceId: string): EventsLogApi { get( eventType: string, levels: ("warn" | "info" | "error" | "debug")[] | "all", - actorId: string, + actorId: string | undefined, filter: EventsLogFilter, limit: number, search?: string ): Promise { return rpc( - `/api/${workspaceId}/log/${eventType}/${actorId}?limit=${limit}${ + `/api/${workspaceId}/log/${eventType}${actorId ? `/${actorId}` : ""}?limit=${limit}${ filter.start ? "&start=" + filter.start.toISOString() : "" }${filter.end ? "&end=" + filter.end.toISOString() : ""}${ levels !== "all" ? `&levels=${levels.join(",")}` : "" diff --git a/webapps/console/pages/[workspaceId]/connections/index.tsx b/webapps/console/pages/[workspaceId]/connections/index.tsx index 6013313a4..164a0bd4b 100644 --- a/webapps/console/pages/[workspaceId]/connections/index.tsx +++ b/webapps/console/pages/[workspaceId]/connections/index.tsx @@ -9,7 +9,7 @@ import { confirmOp, feedbackError, feedbackSuccess } from "../../../lib/ui"; import React, { useState } from "react"; import Link from "next/link"; import { FaExternalLinkAlt, FaPlus, FaTrash } from "react-icons/fa"; -import { index } from "juava"; +import { index, trimMiddle } from "juava"; import { getCoreDestinationType } from "../../../lib/schema/destinations"; import { useRouter } from "next/router"; import { jsonSerializationBase64, useQueryStringState } from "../../../lib/useQueryStringState"; @@ -55,20 +55,31 @@ function EmptyLinks() { ); } +/** + * How to fit a connection title into a narrow space. `"none"` (default) renders the full names and + * lets the container clip them - which cuts the destination off at an arbitrary spot. A number + * trims *each* side to that many characters, in the middle, so both sides stay recognizable + */ +export type TruncationPolicy = "none" | number; + +const truncateTitle = (name: string, policy: TruncationPolicy) => (policy === "none" ? name : trimMiddle(name, policy)); + export const ConnectionTitle: React.FC<{ connectionId: string; stream?: StreamConfig; service?: ServiceConfig; destination: DestinationConfig; showLink?: boolean; -}> = ({ connectionId, stream, service, destination, showLink = false }) => { + truncationPolicy?: TruncationPolicy; +}> = ({ connectionId, stream, service, destination, showLink = false, truncationPolicy = "none" }) => { + const title = (o: { name: string }) => truncateTitle(o.name, truncationPolicy); return (
- {service && } - {stream && } + {service && } + {stream && } {!stream && !service && "DELETED "} {"➞"} - + {showLink && ( = ({ profileBuilder, destination, showLink = false }) => { + truncationPolicy?: TruncationPolicy; +}> = ({ profileBuilder, destination, showLink = false, truncationPolicy = "none" }) => { return (
} size={"small"} // href={stream && link ? `/${stream.workspaceId}/streams?id=${stream?.id}` : undefined} - title={profileBuilder ? profileBuilder.name : "Unknown stream"} + title={profileBuilder ? truncateTitle(profileBuilder.name, truncationPolicy) : "Unknown stream"} /> {destination && ( <> {"→"} - + truncateTitle(d.name, truncationPolicy)} + /> )} {/*{showLink && (*/} diff --git a/webapps/console/pages/api/[workspaceId]/log/[type]/[actorId].ts b/webapps/console/pages/api/[workspaceId]/log/[type]/[actorId].ts index 8b76f725f..820991b35 100644 --- a/webapps/console/pages/api/[workspaceId]/log/[type]/[actorId].ts +++ b/webapps/console/pages/api/[workspaceId]/log/[type]/[actorId].ts @@ -1,138 +1,23 @@ -import { Api, inferUrl, nextJsApiHandler, verifyAccess } from "../../../../../lib/api"; -import { db } from "../../../../../lib/server/db"; +import { Api, inferUrl, nextJsApiHandler } from "../../../../../lib/api"; import { z } from "zod"; -import { getServerLog } from "../../../../../lib/server/log"; -import { ApiError } from "../../../../../lib/shared/errors"; -import { clickhouse } from "../../../../../lib/server/clickhouse"; -import dayjs from "dayjs"; -import utc from "dayjs/plugin/utc"; -import zlib from "zlib"; -import { pipeline } from "node:stream"; -import { getServerEnv } from "../../../../../lib/server/serverEnv"; -dayjs.extend(utc); - -const log = getServerLog("events-log"); -const serverEnv = getServerEnv(); - -//Vercel Limit: https://vercel.com/docs/functions/streaming-functions#limitations-for-streaming-edge-functions -const maxStreamingResponseSize = 100_000_000; +import { eventsLogQuery, streamEventsLog } from "../../../../../lib/server/events-log-stream"; +/** + * Events log of a single actor (site / connection / profile builder). See `./index.ts` for the + * same log across every actor of the workspace + */ export const api: Api = { url: inferUrl(__filename), GET: { types: { - query: z.object({ - type: z.string(), - workspaceId: z.string(), - actorId: z.string(), - levels: z.string().optional(), - limit: z.coerce.number().optional().default(50), - start: z.coerce.date().optional(), - end: z.coerce.date().optional(), - //people can search for ISO timestamps. that we automatically convert to date - search: z.any().optional(), - }), + query: eventsLogQuery.extend({ actorId: z.string() }), result: z.any(), }, streaming: true, auth: true, - handle: async ({ user, req, res, query }) => { - log.atDebug().log("GET", JSON.stringify(query, null, 2)); - await verifyAccess(user, query.workspaceId); - if (query.type === "incoming") { - const source = await db - .prisma() - .configurationObject.findFirst({ where: { id: query.actorId, workspaceId: query.workspaceId } }); - if (!source) { - throw new ApiError(`site doesn't belong to the current workspace`, { status: 403 }); - } - } else { - const link = await db - .prisma() - .configurationObjectLink.findFirst({ where: { id: query.actorId, workspaceId: query.workspaceId } }); - const pb = await db - .prisma() - .profileBuilder.findFirst({ where: { id: query.actorId, workspaceId: query.workspaceId } }); - const dst = await db.prisma().configurationObject.findFirst({ - where: { id: query.actorId, workspaceId: query.workspaceId, type: "destination" }, - }); - if (!link && !pb && !dst) { - throw new ApiError(`connection doesn't belong to the current workspace`, { status: 403 }); - } - } - res.writeHead(200, { - "Content-Type": "application/x-ndjson", - "Content-Encoding": "gzip", - }); - const sqlQuery = `select timestamp as date, level, message as content from events_log - where - actorId = {actorId:String} - and type = {type:String} - ${query.levels ? "and level in ({levels:Array(String)})" : ""} - ${query.start ? "and timestamp >= {start:String}" : ""} - ${query.end ? "and timestamp < {end:String}" : ""} - ${query.search ? "and message ilike concat('%',{search:String},'%')" : ""} - order by timestamp desc limit {limit:UInt32}`; - const chResult = await clickhouse.query({ - query: sqlQuery, - query_params: { - actorId: query.actorId, - type: query.type, - levels: query.levels ? query.levels.split(",") : undefined, - start: query.start ? dayjs(query.start).utc().format("YYYY-MM-DD HH:mm:ss.SSS") : undefined, - end: query.end ? dayjs(query.end).utc().format("YYYY-MM-DD HH:mm:ss.SSS") : undefined, - search: - typeof query.search === "undefined" - ? undefined - : query.search instanceof Date - ? query.search.toISOString() - : query.search, - limit: query.limit, - }, - format: "JSONEachRow", - clickhouse_settings: { - wait_end_of_query: 1, - }, - }); - var responsePromiseResolve; - let responsePromise = new Promise((resolve, reject) => { - responsePromiseResolve = resolve; - }); - const gzip = zlib.createGzip(); - pipeline(gzip, res, err => { - if (err) { - log.atError().withCause(err).log("Error piping data to response"); - } - responsePromiseResolve(); - }); - const stream = chResult.stream(); - stream.on("data", rs => { - for (const r of rs) { - const row = r.json() as any; - if (gzip.bytesWritten < maxStreamingResponseSize) { - const line = JSON.stringify({ - date: dayjs(row.date).utc(true).toDate(), - level: row.level, - content: JSON.parse(row.content), - }); - gzip.write(line + "\n"); - } else { - stream.destroy(); - } - } - }); - stream.on("error", err => { - log.atError().withCause(err).log("Error streaming data"); - gzip.end(); - }); - stream.on("close", () => { - gzip.end(); - }); - stream.on("end", () => { - gzip.end(); - }); - //wait for stream end - await responsePromise; + handle: async ({ user, res, query }) => { + const { actorId, ...rest } = query; + await streamEventsLog({ user, res, query: rest, actorId }); }, }, }; diff --git a/webapps/console/pages/api/[workspaceId]/log/[type]/index.ts b/webapps/console/pages/api/[workspaceId]/log/[type]/index.ts new file mode 100644 index 000000000..dd661a67e --- /dev/null +++ b/webapps/console/pages/api/[workspaceId]/log/[type]/index.ts @@ -0,0 +1,24 @@ +import { Api, inferUrl, nextJsApiHandler } from "../../../../../lib/api"; +import { z } from "zod"; +import { eventsLogQuery, streamEventsLog } from "../../../../../lib/server/events-log-stream"; + +/** + * Events log across every actor of the workspace. Each record carries `actorId`. See + * `./[actorId].ts` for a single actor + */ +export const api: Api = { + url: inferUrl(__filename), + GET: { + types: { + query: eventsLogQuery, + result: z.any(), + }, + streaming: true, + auth: true, + handle: async ({ user, res, query }) => { + await streamEventsLog({ user, res, query }); + }, + }, +}; + +export default nextJsApiHandler(api);