jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768/** the record contract between the strata flow and this worker: one row per sealed stream segment, with the * per-collection event counts decoded from the segment's collection block index. the schema parses the HTTP * body at the boundary; `checkIngestBatch` holds the invariants a schema cannot express. */
import { array, arrayLength, constrain, integer, integerRange, object, optional, string, stringLength, type InferOutput,} from "@atcute/lexicons/validations";
const count = constrain(integer(), [integerRange(0)]);
export const CollectionCountSchema = object({ nsid: constrain(string(), [stringLength(1, 255)]), count,});
export const SegmentRecordSchema = object({ idx: count, name: constrain(string(), [stringLength(1, 64)]), sizeBytes: count, checksum: constrain(string(), [stringLength(1, 32)]), eventCount: count, blockCount: count, minSeq: count, maxSeq: count, minWitnessedAt: count, maxWitnessedAt: count, collections: constrain(array(CollectionCountSchema), [arrayLength(0, 1 << 20)]),});
export const IngestBatchSchema = object({ segments: constrain(array(SegmentRecordSchema), [arrayLength(1, 200)]), /** how many sealed segments the archive reports in total, so the page can say how partial it is */ archiveSegments: optional(count),});
export type CollectionCount = InferOutput<typeof CollectionCountSchema>;export type SegmentRecord = InferOutput<typeof SegmentRecordSchema>;export type IngestBatch = InferOutput<typeof IngestBatchSchema>;
/** the authority prefix of an nsid: its first two labels (`app.bsky.feed.like` → `app.bsky`). */export function namespaceOf(nsid: string): string { const parts = nsid.split("."); return parts.length <= 2 ? nsid : `${parts[0]}.${parts[1]}`;}
/** invariants beyond shape: a segment's seq range runs forward and names each collection once. * returns the first violation as a message, or null when the batch is sound. */export function checkIngestBatch(batch: IngestBatch): string | null { for (const [i, segment] of batch.segments.entries()) { if (segment.maxSeq < segment.minSeq) return `segments[${i}]: maxSeq < minSeq`; const seen = new Set<string>(); for (const c of segment.collections) { if (seen.has(c.nsid)) return `segments[${i}].collections: duplicate nsid ${c.nsid}`; seen.add(c.nsid); } } return null;}