nx_jsonl_writer.nx source
↩ module page · 204 lines · 7607 B
1// nx_jsonl_writer.nx -- streaming JSONL output primitive.
2//
3// module: nishi-core.io.jsonl_writer
4// depends: nishi-core.io.json_emit, nishi-core.io.syscalls
5// disk_kb: 4
6// capability: CORE_IO
7//
8// license_tier: PUBLIC_NISHI_SUBSTRATE
9// genealogy_id: rfc_7464_json_text_sequences +
10// jsonlines_org_jsonl_specification +
11// nishi_catalog_driven_cardinal_2026
12//
13// Streaming line-delimited JSON output. One JSON object per line +
14// newline terminator. Used by EVERY ingestion adapter for its
15// canonical-record emit path (sources/usda_plants emits one row per
16// PlantProfileResult, sources/kew_sid emits one per species, etc.).
17//
18// Per [[feedback-no-pdfs-no-proprietary-binary-formats]]: JSONL is
19// the canonical open-format exchange format for the nishi data
20// pipeline. Per [[feedback-bits-up-canonical-layer-no-reinventing]]:
21// composes against nx_json_emit (shipped) for object serialization.
22//
23// ===== Output discipline ==========================================
24//
25// One JSON object per line. Line terminator = LF (0x0A). NO CRLF.
26// NO leading whitespace. NO trailing whitespace before LF. NO
27// pretty-print -- each record is one compact object. Round-trips
28// through any RFC 7464 / jsonlines.org consumer.
29//
30// File-path convention (per nishi-library staging layout):
31// nishi-library/staging/sources/<source>/<YYYY-MM-DD>/seq-<NNNNNN>.jsonl
32//
33// Each file caps at MAX_RECORDS_PER_FILE records (default 1000).
34// Writer auto-rotates to next seq file when limit hit.
35
36// nx_safety_envelope:
37// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
38// sil_target: SIL1
39// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
40// verdict: NOT_YET_EVALUATED
41
42import "nx_syscalls.nx"
43import "nx_json_emit.nx"
44
45// ===== JsonlWriter struct =========================================
46
47struct JsonlWriter {
48 writer_hk: i64,
49 fd: i64, // file descriptor; -1 if closed
50 record_buf: *u8, // scratch buffer for one record
51 record_buf_cap: i64,
52 record_count: i64, // records emitted to current file
53 file_count: i64, // sequence number (seq-NNNNNN.jsonl)
54 bytes_written: i64, // cumulative
55 max_records_per_file: i64, // rotation threshold
56 err_count: i64, // emit failures
57 last_emit_unix: i64,
58}
59
60const NX_JSONL_WRITER_BYTES: i64 = 80 // 10 fields * 8 bytes
61const NX_JSONL_DEFAULT_BUF_BYTES: i64 = 16384 // 16 KB scratch
62const NX_JSONL_DEFAULT_MAX_RECORDS: i64 = 1000
63
64// ===== Verdict ====================================================
65
66const NX_JSONL_OK: i64 = 1
67const NX_JSONL_FD_INVALID: i64 = 2
68const NX_JSONL_BUFFER_OVERFLOW: i64 = 3
69const NX_JSONL_WRITE_FAIL: i64 = 4
70const NX_JSONL_NOT_OPEN: i64 = 5
71const NX_JSONL_ROTATION_FAILED: i64 = 6
72
73func nx_jsonl_verdict_name(v: i64) -> *u8 {
74 if v == NX_JSONL_OK { return "OK" }
75 if v == NX_JSONL_FD_INVALID { return "FD_INVALID" }
76 if v == NX_JSONL_BUFFER_OVERFLOW { return "BUFFER_OVERFLOW" }
77 if v == NX_JSONL_WRITE_FAIL { return "WRITE_FAIL" }
78 if v == NX_JSONL_NOT_OPEN { return "NOT_OPEN" }
79 if v == NX_JSONL_ROTATION_FAILED { return "ROTATION_FAILED" }
80 return "UNKNOWN"
81}
82
83// ===== Constructor ================================================
84
85func nx_jsonl_writer_new() -> *JsonlWriter {
86 let raw: *u8 = sys_mmap(NX_JSONL_WRITER_BYTES)
87 let w: *JsonlWriter = raw as *JsonlWriter
88 w.writer_hk = 0
89 w.fd = -1
90 w.record_buf = sys_mmap(NX_JSONL_DEFAULT_BUF_BYTES)
91 w.record_buf_cap = NX_JSONL_DEFAULT_BUF_BYTES
92 w.record_count = 0
93 w.file_count = 0
94 w.bytes_written = 0
95 w.max_records_per_file = NX_JSONL_DEFAULT_MAX_RECORDS
96 w.err_count = 0
97 w.last_emit_unix = 0
98 return w
99}
100
101// ===== Bind to file descriptor ====================================
102//
103// Caller opens the file via nx_syscalls (sys_open with O_CREAT|O_WRONLY|
104// O_APPEND) and passes the fd here. Writer owns flush + close.
105
106func nx_jsonl_writer_attach_fd(w: *JsonlWriter, fd: i64) -> i64 {
107 if w == 0 as *JsonlWriter { return NX_JSONL_NOT_OPEN }
108 if fd < 0 { return NX_JSONL_FD_INVALID }
109 w.fd = fd
110 return NX_JSONL_OK
111}
112
113// ===== Emit one record ============================================
114//
115// Caller builds the JSON object via nx_json_emit primitives into
116// w.record_buf, passes the byte length. Writer appends LF and
117// flushes via sys_write. Bumps record_count; auto-rotation flag
118// returned if record_count == max_records_per_file (caller invokes
119// rotate function to swap files).
120
121func nx_jsonl_writer_emit(w: *JsonlWriter, record_len: i64, now_unix: i64) -> i64 {
122 if w == 0 as *JsonlWriter { return NX_JSONL_NOT_OPEN }
123 if w.fd < 0 { return NX_JSONL_NOT_OPEN }
124 if record_len <= 0 { return NX_JSONL_BUFFER_OVERFLOW }
125 if record_len >= w.record_buf_cap { return NX_JSONL_BUFFER_OVERFLOW }
126
127 // Append LF
128 w.record_buf[record_len] = 0x0A
129 let total_len: i64 = record_len + 1
130
131 // Write
132 let n: i64 = sys_write(w.fd, w.record_buf, total_len)
133 if n != total_len {
134 w.err_count = w.err_count + 1
135 return NX_JSONL_WRITE_FAIL
136 }
137 w.record_count = w.record_count + 1
138 w.bytes_written = w.bytes_written + total_len
139 w.last_emit_unix = now_unix
140 return NX_JSONL_OK
141}
142
143// ===== Rotation needed? ===========================================
144//
145// Caller checks after each emit; if 1, caller closes current fd,
146// opens next sequence file, calls nx_jsonl_writer_attach_fd, calls
147// nx_jsonl_writer_mark_rotated.
148
149func nx_jsonl_writer_needs_rotation(w: *JsonlWriter) -> i64 {
150 if w == 0 as *JsonlWriter { return 0 }
151 if w.record_count >= w.max_records_per_file { return 1 }
152 return 0
153}
154
155func nx_jsonl_writer_mark_rotated(w: *JsonlWriter) -> i64 {
156 if w == 0 as *JsonlWriter { return NX_JSONL_NOT_OPEN }
157 w.file_count = w.file_count + 1
158 w.record_count = 0
159 return NX_JSONL_OK
160}
161
162// ===== Build sequence filename ====================================
163//
164// Format: seq-NNNNNN.jsonl (6-digit zero-padded, supports 1M files
165// per source). Caller composes full path:
166// <staging_root>/<source_name>/<YYYY-MM-DD>/<this_filename>
167
168const NX_JSONL_SEQ_FILENAME_LEN: i64 = 16 // "seq-XXXXXX.jsonl"
169
170func nx_jsonl_writer_seq_filename(seq_n: i64, out: *u8, out_cap: i64) -> i64 {
171 if out_cap < NX_JSONL_SEQ_FILENAME_LEN + 1 { return -1 }
172 // "seq-"
173 out[0] = 115 // 's'
174 out[1] = 101 // 'e'
175 out[2] = 113 // 'q'
176 out[3] = 45 // '-'
177 // 6-digit zero-padded number
178 var n: i64 = seq_n
179 out[9] = 0x30 + (n % 10); n = n / 10
180 out[8] = 0x30 + (n % 10); n = n / 10
181 out[7] = 0x30 + (n % 10); n = n / 10
182 out[6] = 0x30 + (n % 10); n = n / 10
183 out[5] = 0x30 + (n % 10); n = n / 10
184 out[4] = 0x30 + (n % 10)
185 // ".jsonl"
186 out[10] = 46 // '.'
187 out[11] = 106 // 'j'
188 out[12] = 115 // 's'
189 out[13] = 111 // 'o'
190 out[14] = 110 // 'n'
191 out[15] = 108 // 'l'
192 out[16] = 0 // NUL
193 return NX_JSONL_SEQ_FILENAME_LEN
194}
195
196// ===== Close ======================================================
197
198func nx_jsonl_writer_close(w: *JsonlWriter) -> i64 {
199 if w == 0 as *JsonlWriter { return NX_JSONL_NOT_OPEN }
200 if w.fd < 0 { return NX_JSONL_NOT_OPEN }
201 sys_close(w.fd)
202 w.fd = -1
203 return NX_JSONL_OK
204}