// std/jsonl - bounded JSON Lines readers and simple writers.
import {
FileAppendCursor,
FileAppendPageOptions,
FileAppendResetReason,
FileLine,
FileLinePageOptions,
FsFailure,
read_lines_append_page_result,
read_lines_page_result,
} from "std/fs"
import { SchemaContract, ValidationIssue, schema_contract_check } from "std/schema"
pub type JsonlErrorHandler = fn(string, JsonlRecordFailure) -> unknown
pub type ReadJsonlOptions = {
strict?: bool,
max_records?: int,
max_bytes?: int,
on_error?: JsonlErrorHandler,
}
type ResolvedReadJsonlOptions = {
strict: bool,
max_records: int,
max_bytes: int,
on_error: JsonlErrorHandler?,
}
pub type JsonlCursor = {offset: int, line: int}
pub type JsonlPageOptions = {cursor?: JsonlCursor, max_records?: int, max_bytes?: int}
pub type JsonlRecord<T> = {line: int, offset: int, raw: string, value: T}
pub type JsonlRecordFailureKind = "malformed" | "schema_invalid" | "rule_failed" | "rule_error"
pub type JsonlRecordFailure = {
kind: JsonlRecordFailureKind,
line: int,
offset: int,
raw: string,
detail: string,
issues?: list<ValidationIssue>,
}
pub type JsonlPage<T> = {
records: list<JsonlRecord<T>>,
issues: list<JsonlRecordFailure>,
next_cursor: JsonlCursor,
done: bool,
}
pub type JsonlPageResult<T> = Result<JsonlPage<T>, FsFailure>
pub type JsonlAppendCursor = FileAppendCursor
pub type JsonlAppendPageOptions = {
cursor?: JsonlAppendCursor,
generation?: string,
max_records?: int,
max_bytes?: int,
}
pub type JsonlAppendPage<T> = {
records: list<JsonlRecord<T>>,
issues: list<JsonlRecordFailure>,
next_cursor: JsonlAppendCursor,
reset_reason?: FileAppendResetReason,
partial_tail_bytes: int,
caught_up: bool,
}
pub type JsonlAppendPageResult<T> = Result<JsonlAppendPage<T>, FsFailure>
fn __read_options(options: ReadJsonlOptions) -> ResolvedReadJsonlOptions {
const opts = options ?? {}
return {
strict: opts.strict ?? false,
max_records: to_int(opts.max_records) ?? 0,
max_bytes: to_int(opts.max_bytes) ?? 1048576,
on_error: opts.on_error,
}
}
fn __page_options(options: JsonlPageOptions) -> FileLinePageOptions {
const opts = options ?? {}
const cursor = opts.cursor ?? {offset: 0, line: 1}
return {
offset: cursor.offset,
line: cursor.line,
max_lines: opts.max_records ?? 256,
max_bytes: opts.max_bytes ?? 1048576,
}
}
fn __append_page_options(options: JsonlAppendPageOptions) -> FileAppendPageOptions {
const opts = options ?? {}
let resolved: FileAppendPageOptions = {
max_lines: opts.max_records ?? 256,
max_bytes: opts.max_bytes ?? 1048576,
}
if opts.cursor != nil {
const cursor = opts.cursor
let file_cursor: FileAppendCursor = {
offset: cursor.offset,
line: cursor.line,
file_identity: cursor.file_identity,
observed_size: cursor.observed_size,
}
if cursor.generation != nil {
file_cursor = file_cursor + {generation: cursor.generation}
}
resolved = resolved + {cursor: file_cursor}
}
if opts.generation != nil {
resolved = resolved + {generation: opts.generation}
}
return resolved
}
fn __malformed(line: FileLine, failure: dict) -> JsonlRecordFailure {
return {
kind: "malformed",
line: line.line,
offset: line.offset,
raw: line.text,
detail: failure?.message ?? to_string(failure),
}
}
fn __parse_line(line: FileLine) -> Result<JsonlRecord<unknown>?, JsonlRecordFailure> {
if trim(line.text) == "" {
return Ok(nil)
}
const parsed = try {
json_parse(line.text)
}
if !is_ok(parsed) {
return Err(__malformed(line, unwrap_err(parsed)))
}
return Ok({line: line.line, offset: line.offset, raw: line.text, value: unwrap(parsed)})
}
/**
* Read one bounded page of JSONL records and preserve malformed rows as issues.
*
* Resume with `page.next_cursor`. Each page holds at most `max_records`
* physical lines and `max_bytes` source bytes. Blank lines advance the cursor
* without producing records.
*
* @effects: [fs.read]
* @errors: []
*/
pub fn read_jsonl_page_result(
fs: HarnessFs,
path: string,
options: JsonlPageOptions = {},
) -> JsonlPageResult<unknown> {
const source = read_lines_page_result(fs, path, __page_options(options))
if !is_ok(source) {
return Err(unwrap_err(source))
}
const lines = unwrap(source)
let records: list<JsonlRecord<unknown>> = []
let issues: list<JsonlRecordFailure> = []
for line in lines.lines {
const parsed = __parse_line(line)
if !is_ok(parsed) {
issues = issues + [unwrap_err(parsed)]
} else if unwrap(parsed) != nil {
records = records + [unwrap(parsed)]
}
}
return Ok(
{
records: records,
issues: issues,
next_cursor: {offset: lines.next_offset, line: lines.next_line},
done: lines.done,
},
)
}
/**
* Incrementally read complete JSONL records from a growing file.
*
* Only newline-terminated physical lines are parsed. An incomplete tail remains
* behind `next_cursor` until a later append commits it. File replacement,
* truncation, and caller-generation changes restart the page at byte zero and
* return an explicit `reset_reason`; malformed committed rows remain issues.
*
* @effects: [fs.read]
* @errors: []
*/
pub fn read_jsonl_append_page_result(
fs: HarnessFs,
path: string,
options: JsonlAppendPageOptions = {},
) -> JsonlAppendPageResult<unknown> {
const source = read_lines_append_page_result(fs, path, __append_page_options(options))
if !is_ok(source) {
return Err(unwrap_err(source))
}
const lines = unwrap(source)
let records: list<JsonlRecord<unknown>> = []
let issues: list<JsonlRecordFailure> = []
for line in lines.lines {
const parsed = __parse_line(line)
if !is_ok(parsed) {
issues = issues + [unwrap_err(parsed)]
} else if unwrap(parsed) != nil {
records = records + [unwrap(parsed)]
}
}
return Ok(
{
records: records,
issues: issues,
next_cursor: lines.next_cursor,
reset_reason: lines.reset_reason,
partial_tail_bytes: lines.partial_tail_bytes,
caught_up: lines.caught_up,
},
)
}
/**
* Throwing form of `read_jsonl_page_result`.
*
* @effects: [fs.read]
* @errors: [FsFailure]
*/
pub fn read_jsonl_page(
fs: HarnessFs,
path: string,
options: JsonlPageOptions = {},
) -> JsonlPage<unknown> {
return unwrap(read_jsonl_page_result(fs, path, options))
}
fn __contract_failure(record: JsonlRecord<unknown>, failure: dict) -> JsonlRecordFailure {
return {
kind: failure.kind,
line: record.line,
offset: record.offset,
raw: record.raw,
detail: failure.detail,
issues: failure.issues,
}
}
/**
* Read one bounded JSONL page and apply a structural schema plus named rules to
* every valid JSON record.
*
* Malformed JSON, structural failures, rule violations, and broken rule
* implementations remain distinct in `page.issues`.
*
* @effects: [fs.read]
* @errors: []
*/
pub fn read_jsonl_contract_page_result<T>(
fs: HarnessFs,
path: string,
contract: SchemaContract<T>,
options: JsonlPageOptions = {},
) -> JsonlPageResult<T> {
const raw = read_jsonl_page_result(fs, path, options)
if !is_ok(raw) {
return Err(unwrap_err(raw))
}
const page = unwrap(raw)
let records: list<JsonlRecord<T>> = []
let issues = page.issues
for record in page.records {
const checked = schema_contract_check(record.value, contract)
if !is_ok(checked) {
issues = issues + [__contract_failure(record, unwrap_err(checked))]
continue
}
records = records
+ [{line: record.line, offset: record.offset, raw: record.raw, value: unwrap(checked)}]
}
return Ok({records: records, issues: issues, next_cursor: page.next_cursor, done: page.done})
}
/**
* Throwing form of `read_jsonl_contract_page_result`.
*
* @effects: [fs.read]
* @errors: [FsFailure]
*/
pub fn read_jsonl_contract_page<T>(
fs: HarnessFs,
path: string,
contract: SchemaContract<T>,
options: JsonlPageOptions = {},
) -> JsonlPage<T> {
return unwrap(read_jsonl_contract_page_result(fs, path, contract, options))
}
/**
* Parse an in-memory JSONL string. Use `read_jsonl_page_result` for files that
* may grow beyond a safe in-memory size.
*
* @effects: []
* @errors: [JsonlRecordFailure]
*/
pub fn parse_jsonl(text: string, options: ReadJsonlOptions = {}) -> list {
return fold_jsonl(text, [], fn(records, record, _index) { return records + [record] }, options)
}
/**
* Materialize a JSONL file for compatibility with list-oriented callers.
*
* The file is still read in bounded pages. Prefer `read_jsonl_page_result` or
* `fold_jsonl_file` when the complete result need not reside in memory.
*
* @effects: [fs.read]
* @errors: [FsFailure, JsonlRecordFailure]
*/
pub fn read_jsonl(fs: HarnessFs, path: string, options: ReadJsonlOptions = {}) -> list {
const opts = __read_options(options)
let out = []
let cursor: JsonlCursor = {offset: 0, line: 1}
while true {
const remaining: int = if opts.max_records > 0 {
opts.max_records - len(out)
} else {
256
}
if remaining <= 0 {
break
}
const page_records: int = min(256, remaining)
const page_options: JsonlPageOptions = {
cursor: cursor,
max_records: page_records,
max_bytes: opts.max_bytes,
}
const page = read_jsonl_page_result(fs, path, page_options)
if !is_ok(page) {
if opts.strict {
throw unwrap_err(page)
}
return out
}
const value = unwrap(page)
for issue in value.issues {
if opts.strict {
throw issue
}
if opts.on_error != nil {
opts.on_error(issue.raw, issue)
}
}
for record in value.records {
out = out + [record.value]
}
cursor = value.next_cursor
if value.done {
break
}
}
return out
}
fn __serialize(items: list) -> string {
let lines = []
for item in items ?? [] {
lines = lines + [json_stringify(item)]
}
return join(lines, "\n") + "\n"
}
/**
* Replace a JSONL file and return the number of records written.
*
* @effects: [fs.write]
* @errors: [FsFailure]
*/
pub fn write_jsonl(fs: HarnessFs, path: string, items: list) -> int {
fs.write_text(path, __serialize(items))
return len(items ?? [])
}
/**
* Append one JSONL record and return the byte count including its newline.
*
* @effects: [fs.write]
* @errors: [FsFailure]
*/
pub fn append_jsonl(fs: HarnessFs, path: string, item: any) -> int {
const line = json_stringify(item) + "\n"
fs.append(path, line)
return len(line)
}
/**
* Fold already-loaded JSONL content without materializing parsed records.
*
* @effects: []
* @errors: [JsonlRecordFailure]
*/
pub fn fold_jsonl<T>(
text: string,
initial: T,
reducer: unknown,
options: ReadJsonlOptions = {},
) -> T {
const opts = __read_options(options)
let state = initial
let index = 0
let line_number = 1
let offset = 0
for text_line in split(text ?? "", "\n") {
if opts.max_records > 0 && index >= opts.max_records {
break
}
const line: FileLine = {line: line_number, offset: offset, text: text_line}
const parsed = __parse_line(line)
if !is_ok(parsed) {
const failure = unwrap_err(parsed)
if opts.strict {
throw failure
}
if opts.on_error != nil {
opts.on_error(line.text, failure)
}
} else if unwrap(parsed) != nil {
state = reducer(state, unwrap(parsed).value, index)
index = index + 1
}
line_number = line_number + 1
offset = offset + len(text_line) + 1
}
return state
}
/**
* Fold a JSONL file through bounded pages.
*
* @effects: [fs.read]
* @errors: [FsFailure, JsonlRecordFailure]
*/
pub fn fold_jsonl_file<T>(
fs: HarnessFs,
path: string,
initial: T,
reducer: unknown,
options: ReadJsonlOptions = {},
) -> T {
const opts = __read_options(options)
let state = initial
let index = 0
let cursor: JsonlCursor = {offset: 0, line: 1}
while true {
const page_options: JsonlPageOptions = {
cursor: cursor,
max_records: 256,
max_bytes: opts.max_bytes,
}
const page = read_jsonl_page_result(fs, path, page_options)
if !is_ok(page) {
throw unwrap_err(page)
}
const value = unwrap(page)
for issue in value.issues {
if opts.strict {
throw issue
}
if opts.on_error != nil {
opts.on_error(issue.raw, issue)
}
}
for record in value.records {
if opts.max_records > 0 && index >= opts.max_records {
return state
}
state = reducer(state, record.value, index)
index = index + 1
}
cursor = value.next_cursor
if value.done {
return state
}
}
}