/** 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;
}