diff --git a/src/persist/_index.yaml b/src/persist/_index.yaml index f471d1f..5eb5dad 100644 --- a/src/persist/_index.yaml +++ b/src/persist/_index.yaml @@ -51,6 +51,7 @@ entries: activation_repo: userspace.dataflow.persist:activation_repo commit_repo: userspace.dataflow.persist:commit_repo dataflow_consts: userspace.dataflow:consts + encoding: userspace.dataflow.persist:encoding ops: userspace.dataflow.persist:ops # userspace.dataflow.persist:commit_repo @@ -68,6 +69,8 @@ entries: - sql - json - time + imports: + encoding: userspace.dataflow.persist:encoding # userspace.dataflow.persist:commit_repo_test - name: commit_repo_test @@ -281,6 +284,7 @@ entries: imports: activation_repo: userspace.dataflow.persist:activation_repo dataflow_consts: userspace.dataflow:consts + encoding: userspace.dataflow.persist:encoding # userspace.dataflow.persist:ops_test - name: ops_test @@ -308,3 +312,23 @@ entries: ops: userspace.dataflow.persist:ops test: wippy.test:test method: run_tests + + # userspace.dataflow.persist:encoding + - name: encoding + kind: library.lua + meta: + comment: The storage boundary — every persisted string is valid UTF-8; invalid sequences become U+FFFD in place. + source: file://encoding.lua + + # userspace.dataflow.persist:encoding_test + - name: encoding_test + kind: function.lua + meta: + type: test + group: Workflow + comment: Invalid external bytes persist as replacement characters instead of failing the commit. + source: file://encoding_test.lua + method: run_tests + imports: + test: wippy.test:test + encoding: userspace.dataflow.persist:encoding diff --git a/src/persist/commit.lua b/src/persist/commit.lua index 31fd3fe..ca0e515 100644 --- a/src/persist/commit.lua +++ b/src/persist/commit.lua @@ -6,6 +6,7 @@ local security = require("security") local ops = require("ops") local commit_repo = require("commit_repo") local consts = require("dataflow_consts") +local encoding = require("encoding") local activation_repo = require("activation_repo") local commit = {} @@ -626,6 +627,9 @@ function commit._create_commit_only(commit_id, dataflow_id, payload, metadata) else payload_json = tostring(payload) end + -- Storage boundary: the commit payload carries node content verbatim and + -- external bytes must not fail the write. + payload_json = encoding.ensure_utf8(payload_json) -- Process metadata - encode tables as JSON or use empty object local metadata_json = "{}" @@ -641,6 +645,7 @@ function commit._create_commit_only(commit_id, dataflow_id, payload, metadata) metadata_json = metadata end end + metadata_json = encoding.ensure_utf8(metadata_json) -- Create timestamp local created_at = time.now():format(time.RFC3339NANO) diff --git a/src/persist/commit_repo.lua b/src/persist/commit_repo.lua index a3ad041..b4f6f5b 100644 --- a/src/persist/commit_repo.lua +++ b/src/persist/commit_repo.lua @@ -1,6 +1,7 @@ local sql = require("sql") local json = require("json") local time = require("time") +local encoding = require("encoding") local DB_RESOURCE = "app:db" @@ -75,6 +76,7 @@ function commit_repo.create(commit_id, dataflow_id, payload, metadata) else payload_json = tostring(payload) end + payload_json = encoding.ensure_utf8(payload_json) -- Process metadata - encode tables as JSON or use empty object local metadata_json = "{}" diff --git a/src/persist/encoding.lua b/src/persist/encoding.lua new file mode 100644 index 0000000..353e1f6 --- /dev/null +++ b/src/persist/encoding.lua @@ -0,0 +1,66 @@ +-- The storage boundary: every string the engine persists is valid UTF-8. +-- External content — scraped pages, model output — carries arbitrary bytes, +-- and Postgres rejects an invalid sequence, which would fail the whole commit +-- and strand the run. Invalid sequences are replaced with U+FFFD in place; +-- the record persists and the damage stays visible and local. +local M = {} + +local REPLACEMENT = "\239\191\189" -- U+FFFD + +function M.ensure_utf8(s: any): any + if type(s) ~= "string" then return s end + local n = #s + local i = 1 + local out: { string }? = nil + local last = 1 + while i <= n do + local c = string.byte(s, i) + local len = 0 + if c < 0x80 then + len = 1 + elseif c >= 0xC2 and c <= 0xDF then + len = 2 + elseif c >= 0xE0 and c <= 0xEF then + len = 3 + elseif c >= 0xF0 and c <= 0xF4 then + len = 4 + end + local ok = len > 0 + if ok and len > 1 then + if i + len - 1 > n then + ok = false + else + for j = 1, len - 1 do + local cc = string.byte(s, i + j) + if cc < 0x80 or cc > 0xBF then + ok = false + break + end + end + if ok and len == 3 then + local c2 = string.byte(s, i + 1) + if (c == 0xE0 and c2 < 0xA0) or (c == 0xED and c2 > 0x9F) then ok = false end + elseif ok and len == 4 then + local c2 = string.byte(s, i + 1) + if (c == 0xF0 and c2 < 0x90) or (c == 0xF4 and c2 > 0x8F) then ok = false end + end + end + end + if ok then + i = i + len + else + if out == nil then out = {} end + local acc = out :: { string } + acc[#acc + 1] = s:sub(last, i - 1) + acc[#acc + 1] = REPLACEMENT + i = i + 1 + last = i + end + end + if out == nil then return s end + local acc = out :: { string } + acc[#acc + 1] = s:sub(last) + return table.concat(acc) +end + +return M diff --git a/src/persist/encoding_test.lua b/src/persist/encoding_test.lua new file mode 100644 index 0000000..ed38e24 --- /dev/null +++ b/src/persist/encoding_test.lua @@ -0,0 +1,49 @@ +local test = require("test") +local encoding = require("encoding") + +local function define_tests() + describe("persist encoding boundary", function() + it("passes valid UTF-8 through untouched", function() + local samples = { + "plain ascii", + "кириллица и ümlaut", + "emoji \240\159\154\128 and CJK \228\184\173\230\150\135", + "", + } + for _, s in ipairs(samples) do + test.eq(encoding.ensure_utf8(s), s) + end + end) + + it("replaces invalid bytes with U+FFFD instead of failing", function() + local dirty = "scraped \255\254 bytes \192\128 inside" + local clean = encoding.ensure_utf8(dirty) + test.is_true(clean:find("\255", 1, true) == nil) + test.is_true(clean:find("\192", 1, true) == nil) + test.is_true(clean:find("scraped ", 1, true) ~= nil) + test.is_true(clean:find(" inside", 1, true) ~= nil) + test.is_true(clean:find("\239\191\189", 1, true) ~= nil) + end) + + it("replaces truncated multi-byte sequences at end of string", function() + local truncated = "tail \226\130" -- first two bytes of a 3-byte sequence + local clean = encoding.ensure_utf8(truncated) + test.eq(clean, "tail \239\191\189\239\191\189") + end) + + it("rejects UTF-16 surrogates encoded as UTF-8", function() + local surrogate = "x\237\160\128y" -- U+D800 + local clean = encoding.ensure_utf8(surrogate) + test.is_true(clean:find("\237", 1, true) == nil) + test.eq(clean:sub(1, 1), "x") + test.eq(clean:sub(-1), "y") + end) + + it("leaves non-strings alone", function() + test.eq(encoding.ensure_utf8(nil), nil) + test.eq(encoding.ensure_utf8(42), 42) + end) + end) +end + +return { run_tests = test.run_cases(define_tests) } diff --git a/src/persist/ops.lua b/src/persist/ops.lua index d049403..c22328b 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -4,6 +4,7 @@ local uuid = require("uuid") local json = require("json") local consts = require("dataflow_consts") local activation_repo = require("activation_repo") +local encoding = require("encoding") -- Use shared constants from consts local constants = { @@ -295,6 +296,7 @@ handlers[constants.COMMAND_TYPES.CREATE_NODE] = function(tx, dataflow_id, op_id, end metadata = encoded end + metadata = encoding.ensure_utf8(metadata) local config = payload.config or "{}" if type(config) == "table" then @@ -552,6 +554,9 @@ handlers[constants.COMMAND_TYPES.CREATE_DATA] = function(tx, dataflow_id, op_id, end content_value = encoded end + -- Storage boundary: external content can carry arbitrary bytes; what is + -- written is always valid UTF-8. + content_value = encoding.ensure_utf8(content_value) local content_type = payload.content_type or "application/json" local node_id = payload.node_id or sql.as.null() @@ -795,12 +800,14 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id, if encode_err then return nil, "Failed to encode content: " .. encode_err end content_value = encoded end + content_value = encoding.ensure_utf8(content_value) local metadata = payload.metadata or "{}" if type(metadata) == "table" then local encoded, encode_err = json.encode(metadata) if encode_err then return nil, "Failed to encode metadata: " .. encode_err end metadata = encoded end + metadata = encoding.ensure_utf8(metadata) local inserted, insert_err = sql.builder.insert("dataflow_data") :set_map({ data_id = payload.data_id,