fix: 修正 logo 中 N 和 G 字母造型
N 添加对角线笔画(█▄ █),G 添加内横杠(█ ▀█), 避免与 O 字母造型雷同。同步更新 ui.ts 中的硬编码 wordmark。
This commit is contained in:
32
packages/stats/server/Dockerfile
Normal file
32
packages/stats/server/Dockerfile
Normal file
@@ -0,0 +1,32 @@
|
||||
FROM oven/bun:1.3.14-alpine AS base
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
ENV NODE_ENV=production
|
||||
ENV BUN_RUNTIME_TRANSPILER_CACHE_PATH=0
|
||||
|
||||
FROM base AS pruner
|
||||
|
||||
COPY . .
|
||||
|
||||
RUN bunx turbo@2.8.13 prune @opencode-ai/stats-server --docker --no-update-notifier --no-color
|
||||
|
||||
FROM base AS installer
|
||||
|
||||
COPY --from=pruner /app/out/json/ ./
|
||||
|
||||
# Bun 1.3.x needs the pruned workspace globs and lockfile metadata refreshed before the frozen production install.
|
||||
RUN bun -e 'const packageJson = await Bun.file("package.json").json(); packageJson.workspaces.packages = Array.from(new Bun.Glob("packages/**/package.json").scanSync(".")).map((file) => file.slice(0, -"/package.json".length)).sort(); await Bun.write("package.json", JSON.stringify(packageJson, null, 2) + "\n")'
|
||||
RUN rm -f bun.lock && bun install --filter @opencode-ai/stats-server --lockfile-only --ignore-scripts
|
||||
RUN bun install --filter @opencode-ai/stats-server --frozen-lockfile --production --ignore-scripts
|
||||
|
||||
FROM base AS runner
|
||||
|
||||
COPY --from=installer /app ./
|
||||
COPY --from=pruner /app/out/full/ ./
|
||||
|
||||
WORKDIR /app/packages/stats/server
|
||||
|
||||
EXPOSE 3000
|
||||
|
||||
CMD ["bun", "src/server.ts"]
|
||||
33
packages/stats/server/package.json
Normal file
33
packages/stats/server/package.json
Normal file
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"$schema": "https://json.schemastore.org/package.json",
|
||||
"name": "@opencode-ai/stats-server",
|
||||
"version": "1.17.4",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"license": "MIT",
|
||||
"main": "./src/server.ts",
|
||||
"exports": {
|
||||
".": "./src/server.ts"
|
||||
},
|
||||
"scripts": {
|
||||
"start": "bun src/server.ts",
|
||||
"typecheck": "tsgo --noEmit"
|
||||
},
|
||||
"dependencies": {
|
||||
"@aws-sdk/client-firehose": "3.933.0",
|
||||
"@effect/platform-node": "catalog:",
|
||||
"@opencode-ai/stats-core": "workspace:*",
|
||||
"effect": "catalog:",
|
||||
"sst": "catalog:"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@tsconfig/node22": "catalog:",
|
||||
"@types/bun": "catalog:",
|
||||
"@types/node": "catalog:",
|
||||
"@typescript/native-preview": "catalog:",
|
||||
"typescript": "catalog:"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=22"
|
||||
}
|
||||
}
|
||||
159
packages/stats/server/src/ingest.ts
Normal file
159
packages/stats/server/src/ingest.ts
Normal file
@@ -0,0 +1,159 @@
|
||||
import { Buffer } from "node:buffer"
|
||||
import { FirehoseClient, PutRecordBatchCommand } from "@aws-sdk/client-firehose"
|
||||
import { Effect, Layer, Schema } from "effect"
|
||||
import * as Context from "effect/Context"
|
||||
import { Resource } from "sst/resource"
|
||||
|
||||
const MAX_FIREHOSE_BATCH_SIZE = 500
|
||||
const MAX_FIREHOSE_ATTEMPTS = 3
|
||||
const LAKE_TYPE = /^([A-Za-z0-9_]+)\.([A-Za-z0-9_]+)$/
|
||||
|
||||
type IngestEvent = Record<string, unknown>
|
||||
type LakeRoute = { database: string; table: string }
|
||||
type FirehoseRecord = { Data: Uint8Array }
|
||||
|
||||
export class IngestError extends Schema.TaggedErrorClass<IngestError>()("IngestError", {
|
||||
message: Schema.String,
|
||||
failed: Schema.Number,
|
||||
cause: Schema.optional(Schema.Defect),
|
||||
}) {}
|
||||
|
||||
export declare namespace Ingest {
|
||||
export interface Service {
|
||||
readonly write: (events: unknown[]) => Effect.Effect<{ records: number }, IngestError>
|
||||
}
|
||||
}
|
||||
|
||||
export class Ingest extends Context.Service<Ingest, Ingest.Service>()("@opencode/stats/Ingest") {
|
||||
static readonly layer: Layer.Layer<Ingest> = Layer.effect(
|
||||
Ingest,
|
||||
Effect.sync(() => {
|
||||
const client = new FirehoseClient({})
|
||||
|
||||
const write = Effect.fn("Ingest.write")(function* (events: unknown[]) {
|
||||
if (events.length === 0) return { records: 0 }
|
||||
const counts = countRoutedEvents(events)
|
||||
if (counts.unsupported > 0) {
|
||||
yield* Effect.logWarning(
|
||||
`lake ingest rejected ${JSON.stringify({ records: counts.records, unsupported: counts.unsupported })}`,
|
||||
)
|
||||
return yield* new IngestError({
|
||||
message: "Unsupported lake event type",
|
||||
failed: counts.unsupported,
|
||||
})
|
||||
}
|
||||
if (counts.records === 0) return { records: 0 }
|
||||
|
||||
let batch: FirehoseRecord[] = []
|
||||
let batches = 0
|
||||
let failed = 0
|
||||
|
||||
for (const event of events) {
|
||||
if (!isRecord(event)) continue
|
||||
const route = routeEvent(event)
|
||||
if (!route) continue
|
||||
batch.push(toFirehoseRecord(event, route))
|
||||
if (batch.length < MAX_FIREHOSE_BATCH_SIZE) continue
|
||||
failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch)
|
||||
batches++
|
||||
batch = []
|
||||
}
|
||||
|
||||
if (batch.length > 0) {
|
||||
failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch)
|
||||
batches++
|
||||
}
|
||||
|
||||
if (failed > 0) {
|
||||
yield* Effect.logWarning(`lake ingest incomplete ${JSON.stringify({ records: counts.records, failed })}`)
|
||||
return yield* new IngestError({ message: "Failed to ingest all lake records", failed })
|
||||
}
|
||||
|
||||
yield* Effect.logInfo(`lake ingest complete ${JSON.stringify({ records: counts.records, batches })}`)
|
||||
return { records: counts.records }
|
||||
})
|
||||
|
||||
return Ingest.of({ write })
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
const putRecords: (
|
||||
client: FirehoseClient,
|
||||
streamName: string,
|
||||
records: FirehoseRecord[],
|
||||
attempt?: number,
|
||||
) => Effect.Effect<number, IngestError> = Effect.fn("Ingest.putRecords")(function* (
|
||||
client,
|
||||
streamName,
|
||||
records,
|
||||
attempt = 1,
|
||||
) {
|
||||
const result = yield* Effect.tryPromise({
|
||||
try: () => client.send(new PutRecordBatchCommand({ DeliveryStreamName: streamName, Records: records })),
|
||||
catch: (cause) =>
|
||||
new IngestError({ message: "Failed to write lake records to Firehose", failed: records.length, cause }),
|
||||
}).pipe(
|
||||
Effect.tapError(() =>
|
||||
Effect.logWarning(`firehose batch write failed ${JSON.stringify({ records: records.length, attempt })}`),
|
||||
),
|
||||
)
|
||||
const failed =
|
||||
result.RequestResponses?.flatMap((item, index) => {
|
||||
const record = records[index]
|
||||
if (!item.ErrorCode || !record) return []
|
||||
return [record]
|
||||
}) ?? []
|
||||
|
||||
if (failed.length === 0) return 0
|
||||
if (attempt >= MAX_FIREHOSE_ATTEMPTS) {
|
||||
yield* Effect.logWarning(
|
||||
`firehose batch failed ${JSON.stringify({ records: failed.length, attempts: MAX_FIREHOSE_ATTEMPTS })}`,
|
||||
)
|
||||
return failed.length
|
||||
}
|
||||
|
||||
yield* Effect.logWarning(
|
||||
`firehose batch retrying ${JSON.stringify({ records: failed.length, attempt: attempt + 1 })}`,
|
||||
)
|
||||
yield* Effect.sleep(`${250 * 2 ** (attempt - 1)} millis`)
|
||||
return yield* putRecords(client, streamName, failed, attempt + 1)
|
||||
})
|
||||
|
||||
function countRoutedEvents(events: unknown[]) {
|
||||
let records = 0
|
||||
let unsupported = 0
|
||||
for (const event of events) {
|
||||
if (!isRecord(event)) continue
|
||||
if (routeEvent(event)) records++
|
||||
else unsupported++
|
||||
}
|
||||
return { records, unsupported }
|
||||
}
|
||||
|
||||
function isRecord(item: unknown): item is IngestEvent {
|
||||
return Boolean(item) && typeof item === "object" && !Array.isArray(item)
|
||||
}
|
||||
|
||||
function routeEvent(event: IngestEvent): LakeRoute | undefined {
|
||||
if (typeof event._datalake_key !== "string") return
|
||||
const match = event._datalake_key.match(LAKE_TYPE)
|
||||
if (!match?.[1] || !match[2]) return
|
||||
return {
|
||||
database: match[1],
|
||||
table: match[2],
|
||||
}
|
||||
}
|
||||
|
||||
function toFirehoseRecord(event: IngestEvent, route: LakeRoute): FirehoseRecord {
|
||||
return {
|
||||
Data: Buffer.from(
|
||||
JSON.stringify({
|
||||
...Object.fromEntries(Object.entries(event).filter(([key]) => key !== "_datalake_key")),
|
||||
_lake_database: route.database,
|
||||
_lake_table: route.table,
|
||||
_lake_operation: "insert" as const,
|
||||
}),
|
||||
),
|
||||
}
|
||||
}
|
||||
11
packages/stats/server/src/resource.d.ts
vendored
Normal file
11
packages/stats/server/src/resource.d.ts
vendored
Normal file
@@ -0,0 +1,11 @@
|
||||
import "sst/resource"
|
||||
|
||||
declare module "sst/resource" {
|
||||
export interface Resource {
|
||||
LakeIngestConfig: {
|
||||
secret: string
|
||||
streamName: string
|
||||
type: "sst.sst.Linkable"
|
||||
}
|
||||
}
|
||||
}
|
||||
73
packages/stats/server/src/router.ts
Normal file
73
packages/stats/server/src/router.ts
Normal file
@@ -0,0 +1,73 @@
|
||||
import { Buffer } from "node:buffer"
|
||||
import { timingSafeEqual } from "node:crypto"
|
||||
import { Effect, Schema } from "effect"
|
||||
import * as Semaphore from "effect/Semaphore"
|
||||
import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
|
||||
import { Resource } from "sst/resource"
|
||||
import { Ingest } from "./ingest"
|
||||
import { isShuttingDown } from "./shutdown"
|
||||
|
||||
const MAX_CONCURRENT_INGEST_REQUESTS = 8
|
||||
|
||||
const IngestPayload = Schema.Struct({
|
||||
events: Schema.optional(Schema.Unknown),
|
||||
})
|
||||
|
||||
export const Routes = HttpRouter.use((router) =>
|
||||
Effect.gen(function* () {
|
||||
const ingestService = yield* Ingest
|
||||
const ingestRequests = yield* Semaphore.make(MAX_CONCURRENT_INGEST_REQUESTS)
|
||||
|
||||
yield* Effect.all(
|
||||
[
|
||||
router.add("GET", "/health", () => json(200, { ok: true })),
|
||||
router.add("GET", "/ready", () => json(isShuttingDown() ? 503 : 200, { ok: !isShuttingDown() })),
|
||||
router.add("POST", "/", ingestRequests.withPermit(ingest(ingestService))),
|
||||
],
|
||||
{ discard: true },
|
||||
)
|
||||
}),
|
||||
)
|
||||
|
||||
const ingest = (ingestService: Ingest.Service) =>
|
||||
Effect.gen(function* () {
|
||||
const request = yield* HttpServerRequest.HttpServerRequest
|
||||
if (!isAuthorized(request.headers)) return yield* json(401, { ok: false, error: "Unauthorized" })
|
||||
|
||||
const payload = yield* HttpServerRequest.schemaBodyJson(IngestPayload).pipe(
|
||||
Effect.match({
|
||||
onFailure: () => undefined,
|
||||
onSuccess: (value) => value,
|
||||
}),
|
||||
)
|
||||
if (!payload) return yield* json(400, { ok: false, error: "Invalid JSON body" })
|
||||
|
||||
const events = Array.isArray(payload.events) ? payload.events : []
|
||||
if (events.length === 0) return yield* json(202, { ok: true, records: 0 })
|
||||
|
||||
return yield* ingestService.write(events).pipe(
|
||||
Effect.flatMap((result) => json(202, { ok: true, records: result.records })),
|
||||
Effect.catchTag("IngestError", (error) =>
|
||||
json(502, { ok: false, records: countRecords(events), failed: error.failed }),
|
||||
),
|
||||
)
|
||||
})
|
||||
|
||||
function isAuthorized(headers: Record<string, string | undefined>) {
|
||||
const actual = Buffer.from(headers.authorization ?? headers.Authorization ?? "")
|
||||
const expected = Buffer.from(`Bearer ${Resource.LakeIngestConfig.secret}`)
|
||||
if (actual.length !== expected.length) return false
|
||||
return timingSafeEqual(actual, expected)
|
||||
}
|
||||
|
||||
function countRecords(items: unknown[]) {
|
||||
let records = 0
|
||||
for (const item of items) {
|
||||
if (Boolean(item) && typeof item === "object" && !Array.isArray(item)) records++
|
||||
}
|
||||
return records
|
||||
}
|
||||
|
||||
function json(status: number, body: Record<string, unknown>) {
|
||||
return HttpServerResponse.json(body, { status }).pipe(Effect.orDie)
|
||||
}
|
||||
28
packages/stats/server/src/server.ts
Normal file
28
packages/stats/server/src/server.ts
Normal file
@@ -0,0 +1,28 @@
|
||||
import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"
|
||||
import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
|
||||
import { Config, Layer } from "effect"
|
||||
import { HttpRouter } from "effect/unstable/http"
|
||||
import { createServer } from "node:http"
|
||||
import { Ingest } from "./ingest"
|
||||
import { Routes } from "./router"
|
||||
import { registerShutdownSignalHandlers } from "./shutdown"
|
||||
|
||||
registerShutdownSignalHandlers()
|
||||
|
||||
const ServerLive = NodeHttpServer.layerConfig(
|
||||
() => createServer(),
|
||||
Config.all({
|
||||
port: Config.number("PORT").pipe(Config.withDefault(3000)),
|
||||
host: Config.string("HOST").pipe(Config.withDefault("0.0.0.0")),
|
||||
}),
|
||||
)
|
||||
|
||||
const runtimeLayer = Ingest.layer
|
||||
const programLayer = Routes.pipe(Layer.provide(runtimeLayer))
|
||||
const main = Layer.launch(
|
||||
HttpRouter.serve(programLayer, {
|
||||
disableLogger: true,
|
||||
}).pipe(Layer.provideMerge(ServerLive)),
|
||||
)
|
||||
|
||||
NodeRuntime.runMain(main, { disableErrorReporting: true })
|
||||
17
packages/stats/server/src/shutdown.ts
Normal file
17
packages/stats/server/src/shutdown.ts
Normal file
@@ -0,0 +1,17 @@
|
||||
let shuttingDown = false
|
||||
let signalHandlersRegistered = false
|
||||
|
||||
export function isShuttingDown() {
|
||||
return shuttingDown
|
||||
}
|
||||
|
||||
export function registerShutdownSignalHandlers() {
|
||||
if (signalHandlersRegistered) return
|
||||
signalHandlersRegistered = true
|
||||
process.once("SIGTERM", markShuttingDown)
|
||||
process.once("SIGINT", markShuttingDown)
|
||||
}
|
||||
|
||||
function markShuttingDown() {
|
||||
shuttingDown = true
|
||||
}
|
||||
22
packages/stats/server/src/stat-sync.ts
Normal file
22
packages/stats/server/src/stat-sync.ts
Normal file
@@ -0,0 +1,22 @@
|
||||
import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
|
||||
import { Athena } from "@opencode-ai/stats-core/athena"
|
||||
import { layer as statsLayer } from "@opencode-ai/stats-core/runtime"
|
||||
import { syncStats } from "@opencode-ai/stats-core/stat-sync"
|
||||
import { Cause, Effect, Layer, Schedule } from "effect"
|
||||
|
||||
const SYNC_INTERVAL = "1 hour"
|
||||
|
||||
const runtimeLayer = Layer.mergeAll(statsLayer, Athena.layer)
|
||||
const syncPass = syncStats().pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.logWarning(`stats sync failed ${JSON.stringify({ cause: Cause.pretty(cause) })}`),
|
||||
),
|
||||
)
|
||||
const daemon = Effect.logInfo("stats sync daemon started").pipe(
|
||||
Effect.andThen(syncPass.pipe(Effect.repeat(Schedule.fixed(SYNC_INTERVAL)))),
|
||||
Effect.forkScoped,
|
||||
)
|
||||
|
||||
NodeRuntime.runMain(Layer.launch(Layer.effectDiscard(daemon).pipe(Layer.provide(runtimeLayer))), {
|
||||
disableErrorReporting: true,
|
||||
})
|
||||
10
packages/stats/server/sst-env.d.ts
vendored
Normal file
10
packages/stats/server/sst-env.d.ts
vendored
Normal file
@@ -0,0 +1,10 @@
|
||||
/* This file is auto-generated by SST. Do not edit. */
|
||||
/* tslint:disable */
|
||||
/* eslint-disable */
|
||||
/* deno-fmt-ignore-file */
|
||||
/* biome-ignore-all lint: auto-generated */
|
||||
|
||||
/// <reference path="../../../sst-env.d.ts" />
|
||||
|
||||
import "sst"
|
||||
export {}
|
||||
12
packages/stats/server/tsconfig.json
Normal file
12
packages/stats/server/tsconfig.json
Normal file
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"$schema": "https://json.schemastore.org/tsconfig",
|
||||
"extends": "@tsconfig/node22/tsconfig.json",
|
||||
"compilerOptions": {
|
||||
"module": "ESNext",
|
||||
"moduleResolution": "bundler",
|
||||
"strict": true,
|
||||
"noEmit": true,
|
||||
"types": ["bun", "node"]
|
||||
},
|
||||
"include": ["src", "../core/src/resource.d.ts"]
|
||||
}
|
||||
Reference in New Issue
Block a user