Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 2x 1x 1x 1x 1x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 9x 1x 10x 10x 10x 10x 10x 10x 10x 10x 10x 26x 26x 26x 26x 26x 10x 10x 10x 10x 5x 5x 5x 5x 5x 5x 5x 2x 3x 1x 1x 10x 9x 4x 1x 1x 1x 4x 4x 4x 10x 10x 10x 10x 3x 3x 3x 10x 10x 6x 1x 1x 1x 6x 6x 6x 10x 10x 10x 1x | import { setTimeout as delay } from "node:timers/promises"
import type { TwitchRuntimeEnv } from "@workspace/env/twitch"
import type { Logger } from "@workspace/logger"
import { TwitchApiService } from "../services/TwitchApiService"
import { TwitchAuthService } from "../services/TwitchAuthService"
import { ConvexService } from "../services/ConvexService"
import { AnnouncementService } from "../services/announcements/AnnouncementService"
import { EventSubRouter } from "../services/eventsub/EventSubRouter"
import { TwitchWebhookServer } from "../services/eventsub/TwitchWebhookServer"
import { GrantStore } from "../auth/grantStore"
import { writeReadiness } from "../runtime/readiness"
import { serializeLogError } from "@workspace/logger"
export class TwitchClient {
readonly auth: TwitchAuthService
readonly webhook: TwitchWebhookServer
private readonly shutdown = new AbortController()
constructor(
private readonly config: TwitchRuntimeEnv,
private readonly logger: Logger,
request: typeof fetch = fetch
) {
const api = new TwitchApiService(config, request, this.shutdown.signal)
this.auth = new TwitchAuthService(
api,
new GrantStore(config.TWITCH_BOT_GRANT_PATH),
config
)
const convex = new ConvexService(
config.CONVEX_URL,
config.TWITCH_WORKER_SECRET,
request,
this.shutdown.signal
)
const announcements = new AnnouncementService(api, this.auth, logger)
const router = new EventSubRouter({ convex, announcements }, logger)
this.webhook = new TwitchWebhookServer(
config.TWITCH_EVENTSUB_SECRET,
convex,
router,
logger
)
}
async run(signal: AbortSignal): Promise<void> {
const abort = () => this.shutdown.abort()
signal.addEventListener("abort", abort, { once: true })
if (signal.aborted) abort()
const identity = {
version: 1 as const,
pid: process.pid,
startedAt: Date.now(),
}
const state = (state: "starting" | "ready" | "unhealthy" | "stopped") =>
writeReadiness(this.config.TWITCH_READINESS_PATH, {
...identity,
state,
updatedAt: Date.now(),
})
try {
await state("starting")
await this.auth.maintain()
await this.webhook.start(this.config.TWITCH_WEBHOOK_PORT)
await this.webhook.recoverPending()
await state("ready")
this.logger.info("Twitch runtime ready")
while (!signal.aborted) {
// Token maintenance and local readiness only. No subscription lifecycle work.
await delay(30000, undefined, { signal })
await this.auth.maintain()
if (!this.webhook.isListening)
throw new Error("Twitch webhook listener unavailable.")
await state("ready")
}
} catch (error) {
if (!signal.aborted) {
await state("unhealthy").catch((failure) =>
this.logger.error("Readiness update failed", {
error: serializeLogError(failure),
})
)
throw error
}
} finally {
await this.webhook
.stop()
.catch((failure) =>
this.logger.error("Webhook shutdown failed", {
error: serializeLogError(failure),
})
)
if (signal.aborted) {
await state("stopped").catch((failure) =>
this.logger.error("Readiness update failed", {
error: serializeLogError(failure),
})
)
this.logger.info("Twitch runtime stopped")
}
signal.removeEventListener("abort", abort)
}
}
}
|