import {
    Observer,
    createDefaultMediasoupRemoteTrackResolverFactory,
    createJsonlFileSinkFactory,
} from '@observertc/observer-js';

const observer = new Observer({
    closeClientIfIdleForMs: 60_000,
    closeCallIfEmptyForMs: 20_000,

    // Publisher ↔ subscriber links, which four detectors and two validators need.
    createRemoteTrackResolver: createDefaultMediasoupRemoteTrackResolverFactory(),

    // Archive every accepted sample, one file per client.
    createClientSink: createJsonlFileSinkFactory({ directory: './stats' }),

    // The one record that outlives the call.
    callSummary: { include: ['clients', 'issues', 'turnServers', 'scores'] },

    // Bake the tenant into appData at birth, resolved once in a middleware.
    createCallAppData:   ({ callId, acceptCtx })      => ({ callId, tenant: acceptCtx?.tenant }),
    createClientAppData: ({ clientId, observedCall }) => ({ clientId, tenant: observedCall.appData.tenant }),
});

// Nothing is implicit — say what you want to watch.
observer
    .addObserverDetector('observer-concurrent-issue-detector', {
        issueTypes: ['congestion', 'ice-disconnected', 'ice-connection-failed'],
        minAffectedCalls: 3,
    })
    .addObserverDetector('client-population-issue-detector', {
        issueTypes: ['cpulimitation'],
        groupBy: 'browser',
    })
    .addObserverDetector('turn-server-health-detector')
    .addObserverDetector('turn-server-outage-detector', { minClientsAtPeak: 10 });

observer.addCallDetector('call-concurrent-issue-detector', {
    issueTypes: ['congestion', 'video-flow-disrupted', 'invented-speech'],
});
observer.addCallDetector('track-delivery-mismatch-detector', {});
observer.addCallDetector('publisher-fault-corroboration-detector', {
    publisherIssueTypes: ['encoder-bottleneck', 'video-capture-bottleneck'],
    receiverIssueTypes: ['video-flow-disrupted', 'dry-inbound-track'],
});

// Prove the resolver is actually wired — the failure mode is permanent silence.
observer.addValidator('remote-track-resolver');

process.on('SIGINT', () => observer.close());

import express from 'express';

const app = express();
app.use(express.json({ limit: '2mb' }));

// Carry the clientId outside the payload so routing never depends on decoding it.
app.post('/api/samples/:clientId', (req, res) => {
    observer.accept(req.body, { receivedAt: Date.now(), tenant: req.header('x-tenant') });
    res.sendStatus(202);
});

import { ClientSampleDecoder } from '@observertc/samples-protobuf-codec';

const decoders = new Map<string, ClientSampleDecoder>();

wss.on('connection', (ws, req) => {
    const clientId = clientIdFrom(req);
    const decoder = new ClientSampleDecoder();
    decoders.set(clientId, decoder);

    ws.on('message', (data: Buffer) => {
        const result = decoder.tryDecode(new Uint8Array(data));
        if (!result.ok) {
            metrics.increment('codec.decode_failed', { code: result.error.code });
            return;
        }
        observer.accept(result.value, { clientId });
    });

    ws.on('close', () => decoders.delete(clientId));
});

import type { AcceptMiddleware } from '@observertc/observer-js';

const route: AcceptMiddleware = ({ sample, context }, next) => {
    sample.callId ??= sample.attachments?.roomId as string;
    sample.clientId ??= sample.attachments?.peerId as string;
    next({ sample, context });
};

const redact: AcceptMiddleware = ({ sample, context }, next) => {
    delete (sample.attachments as Record<string, unknown>)?.email;
    next({ sample, context });
};

observer.addAcceptMiddleware(route, redact);
observer.on('sample-rejected', ({ reason, sample }) => log.warn('dropped', reason, sample.clientId));

observer.on('observer-issue', ({ issue }) => {
    if ((issue.conclusion?.confidence ?? 0) < 0.7) return;

    page({
        title: issue.type,
        summary: issue.conclusion?.summary,
        next: issue.conclusion?.recommendation,
        domain: issue.conclusion?.faultDomain,   // infrastructure | client-population | …
        evidence: issue.payload,
    });
});

observer.on('call-issue', ({ observedCall, issue }) => {
    ticket.attach(observedCall.callId, { type: issue.type, conclusion: issue.conclusion });
});

observer.on('client-issue-resolved', ({ observedClient, resolvedIssue }) => {
    analytics.track('quality_episode', {
        clientId: observedClient.clientId,
        type: resolvedIssue.type,
        durationInMs: resolvedIssue.durationInMs,
        resolvedBy: resolvedIssue.resolvedBy,     // 'client' | 'timeout' | 'client-closed'
    });
});

import { CallHealthAggregator } from '@observertc/observer-js';

observer.on('call-updated', ({ observedCall }) => {
    const health = new CallHealthAggregator(observedCall).aggregate();

    dashboard.push({
        callId: observedCall.callId,
        clients: observedCall.numberOfClients,
        score: observedCall.score,

        degradedRatio: health.degradedRatio,
        inboundDegradedRatio: health.inboundDegradedRatio,
        outboundDegradedRatio: health.outboundDegradedRatio,
        rttMedian: health.rttInMs?.median,
        qualityLimitation: health.qualityLimitation,

        openIssues: [...observedCall.activeIssuesRegistry.values()].map((i) => i.type),
        unconsumed: observedCall.unconsumedOutboundTracks.size,
    });
});

const observer = new Observer({ autoUpdateOnCallUpdate: false });

observer.on('call-added', ({ observedCall }) => {
    // per call, if you also want the call to update only on your tick
    // (ObservedCallSettings.autoUpdateOnClientUpdate: false)
});

setInterval(() => observer.update(), 5_000);

import { Observer } from '@observertc/observer-js';
import { createInterface } from 'node:readline';
import { createReadStream } from 'node:fs';

const replay = new Observer({ autoUpdateOnCallUpdate: true });
replay.addCallDetector('call-concurrent-issue-detector', { issueTypes: ['congestion'] });
replay.on('call-issue', ({ issue }) => console.log(issue.type, issue.conclusion?.summary));

for await (const line of createInterface({ input: createReadStream('./stats/room-1__alice.jsonl') })) {
    replay.accept(JSON.parse(line));
}

replay.close();

observer.on('call-summary', async ({ summary }) => {
    await warehouse.insert('call_summaries', {
        ...summary,
        // truncated is present only when something was really dropped
        trueIssueCount: summary.issues.length + (summary.truncated?.issues ?? 0),
    });
});

const observedRouter = observer.createObservedMediasoupRouter({
    router,
    matchPeerConnectionByWebRtcTransportId: true,
});

observer.on('mediasoup-router-matched-with-peer-connection',
    ({ observedMediasoupRouter, observedCall, observedPeerConnection }) => {
        (observedPeerConnection.appData ??= {}).routerId = observedMediasoupRouter.id;
        store.linkRouterToCall(observedCall.callId, observedMediasoupRouter.id);
    },
);

// For large meetings, sample it yourself — nothing is evicted for you.
setInterval(() => persist(observedRouter.snapshot()), 10_000);