::nx::Class ::nsmcp::AccessRecordStoreSQLite

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables

Class ::nsmcp::AccessRecordStoreSQLite

::nsmcp::AccessRecordStoreSQLite create ... \
           [ -attached (default "false") ] \
           [ -batchSchema (default "") ] \
           [ -db (default "") ] \
           [ -fullScope (default "false") ] \
           [ -handle (default "") ] \
           [ -lastCompleted (default "0") ] \
           [ -lastStamp (default "") ] \
           [ -mode (default "writer") ] \
           [ -observer (default "") ] \
           [ -path path ] \
           [ -peers (default "") ] \
           [ -prior (default "") ] \
           [ -querying (default "false") ] \
           [ -settings:required settings:required ]

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-store.tcl

Class Relations

class: ::nx::Class
superclass: ::nsmcp::Object

::nx::Class create ::nsmcp::AccessRecordStoreSQLite \
     -superclass ::nsmcp::Object

Methods (to be applied on instances)

allScope (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> allScope

Open a query scope covering every active stored generation. Create the source-scope table and retained-record view in a read transaction. This enables complete-generation rollup queries; call finish after querying.

Returns:
An empty result after the all-generation scope is ready.

Testcases:
No testcase defined.

set :fullScope true
if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=OFF}
}
${:db} execute BEGIN
set :querying true
${:db} execute {
    DROP VIEW IF EXISTS temp.nsmcp_record_window; DROP TABLE IF EXISTS temp.nsmcp_scan_batch; DROP TABLE IF
    EXISTS temp.nsmcp_record_scope;
    CREATE TEMP TABLE nsmcp_record_scope(fid INTEGER PRIMARY KEY,lo INTEGER,hi INTEGER);
    INSERT INTO temp.nsmcp_record_scope SELECT fid,0,high FROM nsmcp_records.files WHERE active=1
}
set floor [:retentionFloor]
${:db} execute "
    CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM nsmcp_records.rows r JOIN temp.nsmcp_record_scope s
    ON r.fid=s.fid AND r.off>=s.lo AND r.stop<=s.hi WHERE r.kind<>0 OR r.t>=$floor
"
if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=ON}
}

available (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> available
```
Report whether this adapter owns a database session. This checks the current object state; it does not obtain another pool handle or perform a database health query.

Returns:
True when a session is available, otherwise false.

Testcases:
No testcase defined.
```
return [expr {${:db} ne {}}]
```
batchInsertionAvailable (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> batchInsertionAvailable
```
Report use of the prepared-row batch insertion API. Current deployments require this API. Tests may override this method to select the SQL reference implementation for comparison.

Returns:
True for the required production insertion path.

Testcases:
No testcase defined.
```
return true
```

cRead (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> cRead fid file \
    offset stop budget limits redactor ingestOnly [ batched ]

Read a bounded source slice through the C access-log scanner. Use a connection-local temporary table to materialize raw scanner fields, then cook and redact valid requests. Raw fields are never written to the canonical store. Leave an incomplete tail uncommitted.

Parameters:

fid (required)
Stored source-file identifier.

file (required)
Discovered source descriptor.

offset (required)
Starting source byte offset.

stop (required)
Planned exclusive source boundary.

budget (required)
Maximum source bytes to scan in this invocation.

limits (required)
Scanner limits, including the maximum line size.

redactor (required)
Redaction object used when cooking requests.

ingestOnly (required)
True for ingestion; false allows time-dependent records to request the Tcl scanner fallback.

batched (optional, defaults to "false")
True to construct prepared rows; false to construct SQL reference tuples.

Returns:
A list of end offset, cooked tuples, pending flag, fallback flag and parsing-loss counts.

Testcases:
No testcase defined.

:accessIngestionObserve batch-capabilities
if {![llength [info commands ns_accesslog]]} {
    error "C scanner ingestion requires nslog's C scan API"
}
set mapping {
    offset off
    endOffset stop
    timestamp stamp
    peerAddress peer
    threadName identifier
    user usr
    requestLine request
    status status
    responseBytes bytes
    userAgent agent
    startTime start
    acceptTime a
    queueTime q
    filterTime f
    runTime r
    parseStatus parsed
    diagnostic diagnostic
}
:accessIngestionObserve prepare-raw-stage
${:db} execute {
    DROP TABLE IF EXISTS temp.nsmcp_scan_batch;
    CREATE TEMP TABLE nsmcp_scan_batch(off INTEGER, stop INTEGER, stamp TEXT, peer TEXT, identifier TEXT, usr
                                       TEXT, request TEXT, status INTEGER, bytes INTEGER, agent TEXT, start INTEGER, a INTEGER, q INTEGER, f
                                       INTEGER, r INTEGER, parsed TEXT, diagnostic TEXT)
}
set timing [::nsmcp::accessRecordParser cget -timingColumn]
set timing [dict get {-2 none -1 auto 0 0 2 2} $timing]
set scanOptions [list -filename [dict get $file path] -offset $offset -endoffset $stop -maxlinebytes  [dict get $limits line]]
if {$timing ne "auto"} {
    lappend scanOptions -timingcolumn $timing
}
:accessIngestionObserve open-c-scanner
set scanner [ns_accesslog scan open {*}$scanOptions]
set begin $offset
set tuples {}
set pending false
set counts {malformed 0 oversized 0 skipped 0}
try {
    # The scanner enforces this invocation's planned end even if the
    # active log grew after discovery or this is a covered-range gap.
    # Keep framing state within this call, but serialize only completed
    # record offsets between worker slices. Never commit a snapshot tail.
    :accessIngestionObserve scan-c
    set batch [ns_accesslog scan next $scanner -maxrecords 512 -maxscanbytes  [expr {min(1073741824,max(1,$budget))}] -maxbytes  [expr {max(8388608,4*[dict get $limits line]+2048)}] -timeout 100ms]
    set info [ns_db batchinfo $batch]
    :accessIngestionObserve stage-raw-batch
    ${:db} transaction {
        ${:db} batchInsert -mapping $mapping {temp nsmcp_scan_batch} $batch
    }
    :accessIngestionObserve cook-records
    foreach item [${:db} execute  {SELECT *,agent IS NOT NULL AS combined FROM temp.nsmcp_scan_batch ORDER BY off}] {
        set next [dict get $item stop]
        if {$next>$stop} {
            set pending true
            break
        }
        set kind 0
        set row {}
        if {[dict get $item parsed] ne "ok"} {
            set kind [expr {[dict get $item diagnostic] eq "oversized" ? 2 : 1}]
        } else {
            # Consume decoded C fields and typed timing values directly.
            set timing [::nsmcp::accessRecordParser timingMicroseconds [dict get $item stamp]  [dict get $item start] [dict get $item a] [dict get $item q]  [dict get $item f] [dict get $item r]]
            set bytes [dict get $item bytes]
            if {$bytes eq {}} {
                set bytes 0
            }
            if {$timing eq {}} {
                set row {}
            } else {
                set row [::nsmcp::accessRecordParser requestFields [dict get $item peer]  [dict get $item identifier] [dict get $item usr] [dict get $item stamp]  [dict get $item request] [dict get $item status] $bytes  [dict get $item combined] [dict get $item agent] $timing]
            }
            if {$row ne {} && [dict exists $row timeDependent] && !$ingestOnly} {
                return [list $begin {} false true {malformed 0 oversized 0 skipped 0}]
            }
            if {$row eq {} || [dict exists $row timeDependent]} {
                set row {}
                set kind 1
            }
        }
        if {$kind==0} {
            lappend tuples [:tuple $fid $offset $next 0 $row $redactor $file $batched]
        } elseif {$kind==2} {
            dict incr counts oversized
        } else {
            dict incr counts malformed
        }
        set offset $next
    }
    if {[dict get $info state] eq "pending"} {
        set pending true
    }
} finally {
    ns_accesslog scan close $scanner
    ${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_scan_batch}
}
return [list $offset $tuples $pending false $counts]

capacity (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> capacity

Report the configured record-store budget and current allocation. Account for the main database and its journal files separately. Writer instances also publish a timestamped capacity snapshot for the status interface.

Returns:
A dictionary containing byte budgets, allocation, disk usage and the limit-exceeded flag.

Testcases:
No testcase defined.

set pageSize [${:db} scalar {PRAGMA nsmcp_records.page_size}]
set allocated [expr {wide([${:db} scalar {PRAGMA nsmcp_records.page_count}])*$pageSize}]
set mainLimit [expr {[:accessRecordMainPages ${:settings} $pageSize]*$pageSize}]
set disk [:diskBytes]
set budget [dict get ${:settings} maxBytes]
set result [dict create  maxBytes $budget  mainMaxBytes $mainLimit  mainAllocatedBytes $allocated  diskBytes $disk  limitExceeded [expr {$allocated>$mainLimit || $disk>$budget}]  minimumBudgetBytes [expr {($allocated*5+1)/2}]]
if {${:mode} eq "writer"} {
    nsv_set nsmcp:access-store:[ns_info server] capacitySnapshot  [dict merge $result [dict create observedAt [clock seconds]]]
}
return $result

close (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> close
```
Destroy this adapter and release its database resources. Roll back any unfinished query or transaction, remove temporary state, detach the record store and restore the connection settings before returning its handle to the pool.

Returns:
No result; the adapter object is destroyed.

Testcases:
No testcase defined.
```
:destroy
```
coveredBytes (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> coveredBytes fid lo \
    hi
```
Count committed bytes overlapping a source interval. Intersect the interval with persisted coverage ranges. This is stored coverage, not a measurement of current source-file contents.

Parameters:

fid (required)
Stored source-file identifier.

lo (required)
Inclusive lower byte boundary.

hi (required)
Exclusive upper byte boundary.

Returns:
The number of covered bytes in the interval.

Testcases:
No testcase defined.
```
return [${:db} scalar {
    SELECT coalesce(sum(min(hi, :hi)-max(lo, :lo)), 0) FROM nsmcp_records.ranges WHERE fid=:fid AND hi>:lo AND
    lo<:hi
}]
```
coveredEnd (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> coveredEnd fid \
    offset end
```
Find the end of the committed range containing an offset. Use persisted byte coverage rather than reading the source file. Clamp the result to the caller's planned scan boundary.

Parameters:

fid (required)
Stored source-file identifier.

offset (required)
Starting byte offset to test.

end (required)
Upper byte boundary for this scan.

Returns:
The covered end offset, or offset itself when no range covers it.

Testcases:
No testcase defined.
```
set hi [${:db} scalar {
    SELECT coalesce(max(hi),-1) FROM nsmcp_records.ranges WHERE fid=:fid AND lo<=:offset AND hi>:offset
}]
return [expr {$hi<0 ? $offset : min($hi,$end)}]
```

descriptors (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> descriptors

Describe the active generations present in the record store. Derive live and rotated-source descriptors from persisted state without discovering or opening source files.

Returns:
A list of active stored-source descriptors and their committed byte boundaries.

Testcases:
No testcase defined.

set result {}
set i 0
set control [:stateGet continuous {}]
foreach row [${:db} execute  {SELECT fid,identity,high FROM nsmcp_records.files WHERE active=1 ORDER BY fid}] {
    set fid [dict get $row fid]
    set live [expr {[dict exists $control live $fid]}]
    lappend result [dict create  archiveFid $fid  identity [dict get $row identity]  fileIndex $i  source 0  rotation [expr {!$live}]  compressed false  size [dict get $row high]  offset [dict get $row high]  status complete]
    incr i
}
return $result

diagnostic (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> diagnostic message \
    opts
```
Classify and log a record-store failure. Use the shared aggregation diagnostic path so callers receive a bounded classification and diagnostic identifier rather than exposing the underlying error details.

Parameters:

message (required)
Original error message.

opts (required)
Tcl return-options dictionary for the failure.

Returns:
The shared diagnostic dictionary for the record-store error.

Testcases:
No testcase defined.
```
return [accessAggregationDiagnostic record-store $message $opts ${:handle}]
```
diskBytes (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> diskBytes
```
Measure the disk space occupied by the record-store files. Include the database, rollback journal, WAL and shared-memory file when present. SQLite temporary files and unrelated files are excluded.

Returns:
The total size in bytes of the existing store files.

Testcases:
No testcase defined.
```
set total 0
foreach suffix {{} -journal -wal -shm} {
    if {[file exists ${:path}$suffix]} {
        incr total [file size ${:path}$suffix]
    }
}
return $total
```
dropState (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> dropState fid \
    offset
```
Check whether an offset follows a committed dropped section. Consult the source generation's persisted dropping flag and high watermark. This allows the scanner to carry its framing state across ingestion slices.

Parameters:

fid (required)
Stored source-file identifier.

offset (required)
Byte offset at which scanning will resume.

Returns:
True when the generation is dropping at or before this offset, otherwise false.

Testcases:
No testcase defined.
```
return [${:db} scalar {
    SELECT CASE WHEN dropping=1 AND high<=:offset THEN 1 ELSE 0 END FROM nsmcp_records.files WHERE fid=:fid
}]
```
finish (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> finish
```
Finish the current scoped read transaction. Commit the transaction and clear the object's query-active flag. The adapter continues to own its session until it is closed.

Returns:
False after the query-active flag has been cleared.

Testcases:
No testcase defined.
```
${:db} execute COMMIT
set :querying false
```
gapEnd (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> gapEnd fid offset \
    end
```
Find the next committed range after an uncovered offset. Bound scanning to the next stored range or the planned source end, whichever comes first.

Parameters:

fid (required)
Stored source-file identifier.

offset (required)
Uncovered byte offset.

end (required)
Upper byte boundary for this scan.

Returns:
The end offset of the uncovered interval.

Testcases:
No testcase defined.
```
set lo [${:db} scalar  {SELECT coalesce(min(lo),-1) FROM nsmcp_records.ranges WHERE fid=:fid AND lo>:offset}]
return [expr {$lo<0 ? $end : min($lo,$end)}]
```

ingest (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> ingest fid \
    checkpoint tuples [ batched ]

Persist the scanner's supplied requests and checkpoint atomically. Insert every supplied request; duplicate source positions are errors. Commit rollups, pending peer addresses, parsing-loss counters and byte coverage in the same transaction. This method does not read the source file.

Parameters:

fid (required)
Stored source-file identifier.

checkpoint (required)
Scanner-prepared coverage, framing, timestamp and parsing-loss metadata.

tuples (required)
Prepared request rows, or serialized SQL tuples for the reference path.

batched (optional, defaults to "false")
True for the batch API; false for the SQL reference implementation.

Returns:
Committed request and diagnostic counts, together with the insertion-backend label.

Testcases:
No testcase defined.

foreach key {begin end lo hi anchor stamp now dropping} {
    set $key [dict get $checkpoint $key]
}
set malformed [:dict getdef $checkpoint malformed 0]
set oversized [:dict getdef $checkpoint oversized 0]
set skipped [:dict getdef $checkpoint skipped 0]
foreach count [list $malformed $oversized $skipped] {
    if {![string is entier -strict $count] || $count<0} {
        error "invalid checkpoint counter"
    }
}
set committed [dict create  records 0  requestRecords 0  malformedRecords $malformed  oversizedRecords $oversized]
if {$batched && [llength $tuples] > 512} {
    error "record-store-batch-limit"
}
:accessIngestionObserve begin-transaction
${:db} transaction {
    if {![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=:fid}]} {
        error "record-store-generation-changed"
    }
    if {[llength $tuples]} {
        set previousRowid  [${:db} scalar {SELECT coalesce(max(rowid),0) FROM nsmcp_records.rows}]
        if {$previousRowid == 9223372036854775807} {
            error "record-store-rowid-limit"
        }
        :accessIngestionObserve insert-records
        if {$batched} {
            ${:db} batchInsert -schema ${:batchSchema} {nsmcp_records rows} $tuples
        } else {
            ${:db} executeDml "INSERT INTO nsmcp_records.rows VALUES [join $tuples ,]"
        }
        # The writer lock makes this watermark identify precisely the
        # inserted batch, without staging or filtering its records.
        set batchSource "(SELECT * FROM nsmcp_records.rows WHERE rowid>$previousRowid)"
        :accessIngestionObserve count-records
        set committed [dict merge $committed [lindex [${:db} execute "
            SELECT count(*) AS records,
            coalesce(sum(kind=0),0) AS requestRecords FROM $batchSource
        "] 0]]
        :accessIngestionObserve rollup
        :rollBatch $batchSource
        :accessIngestionObserve observe-records
        :observeBatch $batchSource
    }
    :accessIngestionObserve store-peers
    if {[dict size ${:peers}]} {
        set values {}
        dict for {key data} ${:peers} {
            lassign $data address seen
            lappend values [list $key $address $seen]
        }
        ${:db} executeDml  -bindtypes {values {rows {binary scalar scalar}}} {
                INSERT INTO nsmcp_records.peer_addresses VALUES :values
                ON CONFLICT(k) DO UPDATE SET address=excluded.address,seen=excluded.seen
                WHERE excluded.seen>=seen AND (excluded.seen<>seen OR excluded.address<>address)
            }
    }
    :accessIngestionObserve checkpoint-counters
    # Parsing losses are committed with the same source interval. A
    # bad-only batch advances coverage without inserting a request row.
    if {$malformed+$oversized+$skipped>0} {
        set observed  [:stateGet observations {first 0 latest 0 completed 0 malformed 0 oversized 0}]
        dict incr observed malformed $malformed
        dict incr observed oversized $oversized
        :statePut observations $observed
        ${:db} execute {
            INSERT INTO nsmcp_records.scan_counts VALUES(:fid,:begin,:end,:malformed,:oversized,:skipped,:now)
        }
    }
    dict incr committed records [expr {$malformed + $oversized + $skipped}]
    :accessIngestionObserve save-ranges
    ${:db} execute {
        DELETE FROM nsmcp_records.ranges WHERE fid=:fid AND lo<=:end AND hi>=:begin;
        INSERT INTO nsmcp_records.ranges VALUES(:fid,:lo,:hi,:anchor);
        UPDATE nsmcp_records.files
        SET dropping=CASE WHEN :end>=high THEN :dropping ELSE dropping END,
        stamp=CASE WHEN :end>=high THEN :stamp ELSE stamp END,
        high=max(high,:end),touched=:now
        WHERE fid=:fid
    }
    :accessIngestionObserve check-storage
    if {[:diskBytes] > [dict get ${:settings} maxBytes]} {
        error "record-store-disk-limit"
    }
    :accessIngestionObserve commit
}
set :peers {}
dict set committed insertionBackend [expr {$batched ? "batch-api" : "sql-values"}]
return $committed

legacyViews (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> legacyViews

Create connection-local views for the access-analytics interface. Expose canonical requests, control metadata and evictions through the legacy view names. Enforce request and peer retention without duplicating stored records.

Returns:
An empty result after the temporary views have been created.

Testcases:
No testcase defined.

if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=OFF}
}
set floor [:retentionFloor]
set peerFloor [expr {[clock seconds]-[dict get ${:settings} peerRetention]}]
${:db} execute {
    DROP VIEW IF EXISTS temp.nsmcp_access_requests; DROP VIEW IF EXISTS temp.nsmcp_access_metadata; DROP VIEW IF
    EXISTS temp.nsmcp_access_evictions
}
${:db} execute "
    CREATE TEMP VIEW nsmcp_access_requests AS SELECT r.fid AS generation, r.off AS offset, r.t AS start,
    r.completed AS completed, r.identifier AS request_id,
    CASE WHEN length(r.path)>512 THEN '' ELSE r.path END AS path, r.pool, r.method, r.status, r.bytes,
    r.fallback, (length(r.path)>512) AS excluded,
    CASE WHEN r.phase THEN r.a ELSE NULL END AS accept,CASE WHEN r.phase THEN r.q ELSE NULL END AS queue,
    CASE WHEN r.phase THEN r.f ELSE NULL END AS filter,CASE WHEN r.phase THEN r.r ELSE NULL END AS run,
    CASE WHEN r.t>=$peerFloor THEN coalesce(p.address, '') ELSE '' END AS peer, r.auth AS authenticationState,
    r.label AS userAgent
    FROM nsmcp_records.rows r JOIN nsmcp_records.files f ON r.fid=f.fid AND f.active=1
    LEFT JOIN nsmcp_records.peer_addresses p ON r.peer=p.k AND p.seen>=$peerFloor WHERE r.kind=0 AND r.t>=$floor
"
${:db} execute {
    CREATE TEMP VIEW nsmcp_access_metadata AS SELECT key,value FROM nsmcp_records.state;
    CREATE TEMP VIEW nsmcp_access_evictions AS SELECT * FROM nsmcp_records.evictions
}
if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=ON}
}

observeBatch (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> observeBatch \
    [ batchSource ]

Update observation metadata from an inserted or migrated batch. Merge first and latest request times, completion time and parsing-loss counts into persistent state. The caller supplies the surrounding transaction.

Parameters:

batchSource (optional, defaults to "temp.nsmcp_ingest_batch")
Trusted SQL relation containing the batch; defaults to the migration staging table.

Returns:
An empty result after the observation state has been saved.

Testcases:
No testcase defined.

set row [lindex [${:db} execute "
    SELECT coalesce(min(CASE WHEN kind=0 THEN t END), 0) AS first, coalesce(max(t), 0) AS latest,
    coalesce(max(CASE WHEN kind=0 THEN completed END), 0) AS completed, coalesce(sum(kind=1), 0) AS malformed,
    coalesce(sum(kind=2), 0) AS oversized FROM $batchSource
"] 0]
set old [:stateGet observations {first 0 latest 0 completed 0 malformed 0 oversized 0}]
set first [dict get $row first]
if {$first>0 && ([dict get $old first]==0 || $first<[dict get $old first])} {
    dict set old first $first
}
foreach key {latest completed} {
    dict set old $key [expr {max([dict get $old $key],[dict get $row $key])}]
}
foreach key {malformed oversized} {
    dict incr old $key [dict get $row $key]
}
:statePut observations $old

overview (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> overview

Report persisted ingestion progress and current store status. Combine the continuous cursor, observation times, parsing-loss totals, retention floor and capacity. Missing history metadata remains pending rather than implying complete coverage.

Returns:
An overview dictionary containing coverage, lag, storage, writer and retention status.

Testcases:
No testcase defined.

set report [:stateGet continuous {}]
set observed [:stateGet observations {}]
set floor [:retentionFloor]
set now [clock seconds]
foreach key {first latest completed} {
    set $key [:dict getdef $observed $key 0]
    if {!([set $key] > 0)} {
        set $key {}
    }
}
set size [:dict getdef $report observedSize 0]
set cursor [:dict getdef $report cursor 0]
set history [:dict getdef $report historyUnreadBytes 1]
set unsupported [:dict getdef $report unsupportedCompressed 0]
if {$unsupported} {
    incr history $unsupported
}
set recent [:dict getdef $report recentFrom $now]
set updated [:dict getdef $report updatedAt 0]
return [dict create  observedSize $size  cursor $cursor  historyEnd $history  historyCursor 0  recentFrom $recent  initialOffset 0  recentInitialized [expr {$updated>0}]  seek {}  updatedAt $updated  firstObserved $first  latestObserved $latest  latestCompletionObserved $completed  pending [:dict getdef $report pending 0]  generation 1  rotationGaps 0  lastRotationAt 0  malformed  [:dict getdef $observed malformed 0]  oversized  [:dict getdef $observed oversized 0]  lastMalformedAt {}  ingestionLagSeconds [expr {
                                       [:dict getdef $report caughtUp false] ? max(0, $now-$updated) :
                                       ($completed eq {} ? max(0, $now-$updated) : max(0, $now-$completed))
                                   }]  retentionFloor $floor  storage [:capacity]  writer [accessIngestionStatus]  retention [accessRetentionStatus]]

prepare (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> prepare file

Resolve or initialize the stored generation for a discovered log file. Validate the source prefix and committed-range anchors before reusing a generation. Writers retire or replace changed generations; readers only accept an existing active generation.

Parameters:

file (required)
Discovered source descriptor, including path, identity, size and modification time.

Returns:
The stored file identifier, or -1 when a reader cannot use the generation.

Testcases:
No testcase defined.

file stat [dict get $file path] actual
set identity [dict get $file identity]
set policy [:dict getdef ${:settings} policy {}]
set key [ns_crypto::md string -digest sha256 -- [list schema1 $identity $policy]]
set fid {}
set old [${:db} execute {SELECT * FROM nsmcp_records.files WHERE k=:key}]
set f [open [dict get $file path] rb]
try {
    set reset false
    if {$old ne {}} {
        set old [lindex $old 0]
        set fid [dict get $old fid]
        set prefix  [ns_crypto::md string -digest sha256 -binary -- [read $f [dict get $old plen]]]
        if {[dict get $old high]>$actual(size) || $prefix ne [dict get $old prefix]
            || ([dict get $old high]==$actual(size) && [dict get $old stamp]!=$actual(mtime))} {
            set reset true
        }
        if {!$reset} {
            # A small boundary anchor per contiguous imported range
            # detects rewrites near committed cursors before reuse.
            foreach span [${:db} execute  {SELECT hi,anchor FROM nsmcp_records.ranges WHERE fid=:fid LIMIT 4097}] {
                set hi [dict get $span hi]
                seek $f [expr {max(0,$hi-64)}]
                if {[ns_crypto::md string -digest sha256 -binary --  [read $f [expr {min(64,$hi)}]]] ne [dict get $span anchor]} {
                    set reset true
                    break
                }
            }
        }
    }
    if {${:mode} eq "reader"} {
        return [expr {$old eq {} || $reset || ![dict get $old active] ? -1 : $fid}]
    }
    if {$old eq {} || $reset} {
        set plen [expr {min(64,[dict get $file size])}]
        seek $f 0
        set prefix [ns_crypto::md string -digest sha256 -binary -- [read $f $plen]]
        set high 0
        set stamp [dict get $file mtime]
        set now [clock seconds]
        ${:db} transaction {
            if {[:dict getdef ${:settings} unified false] && $fid ne {}} {
                set retired "${key}:retired:$fid"
                set oldIdentity "${identity}:retired:$fid"
                ${:db} execute {
                    UPDATE nsmcp_records.files SET k=:retired,identity=:oldIdentity WHERE fid=:fid
                }
            } else {
                ${:db} execute {
                    DELETE FROM nsmcp_records.rows WHERE fid=:fid; DELETE FROM nsmcp_records.ranges WHERE
                    fid=:fid; DELETE FROM nsmcp_records.files WHERE k=:key
                }
            }
            ${:db} execute {
                INSERT INTO nsmcp_records.files(k, identity, prefix, plen, high, stamp, touched) VALUES(:key,
                                                                                                        :identity, :prefix, :plen, :high, :stamp, :now)
            }
        }
    }
} finally {
    close $f
}
set fid [${:db} scalar {SELECT fid FROM nsmcp_records.files WHERE k=:key}]
set now [clock seconds]
set touchBefore [expr {$now-max(1,min(3600,int([dict get ${:settings} retention]/4)))}]
${:db} execute  {UPDATE nsmcp_records.files SET touched=:now WHERE fid=:fid AND touched<:touchBefore}
${:db} execute {
    UPDATE nsmcp_records.files SET active=0 WHERE identity=:identity AND fid<>:fid AND active=1;
    UPDATE nsmcp_records.files SET active=1 WHERE fid=:fid
}
return $fid

prune (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> prune [ batchRows ]

Apply the configured retention policy to a writable store. Unified stores remove bounded batches of expired records and related metadata. Other stores expire whole source generations so retained coverage cannot certify deleted requests.

Parameters:

batchRows (optional, defaults to "10000")
Maximum rows per unified-retention deletion, from 1 through 10000.

Returns:
Deletion counts for unified retention, or an empty result for generation retention.

Testcases:
No testcase defined.

if {[:dict getdef ${:settings} unified false]} {
    return [:pruneAge $batchRows]
}
# Expire whole source generations, not individual rows/ranges: keeping
# a range after deleting its rows would falsely certify cached coverage.
set cutoff [expr {[clock seconds]-[dict get ${:settings} retention]}]
${:db} transaction {
    ${:db} execute {
        DELETE FROM nsmcp_records.rows WHERE fid IN (SELECT fid FROM nsmcp_records.files WHERE touched<:cutoff);
        DELETE FROM nsmcp_records.ranges WHERE fid IN (SELECT fid FROM nsmcp_records.files WHERE
                                                       touched<:cutoff);
        DELETE FROM nsmcp_records.scan_counts WHERE fid IN (SELECT fid FROM nsmcp_records.files WHERE
                                                            touched<:cutoff);
        DELETE FROM nsmcp_records.files WHERE touched<:cutoff
    }
}

rollupBackfill (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> rollupBackfill \
    [ batches ]

Build missing minute rollups for previously stored requests. Process bounded batches in source order. Commit rollups, observations, rolled flags and the migration cursor atomically for each batch.

Parameters:

batches (optional, defaults to "1")
Maximum number of migration batches to process in this invocation.

Returns:
True when no unrolled rows remain; false when the batch budget was exhausted.

Testcases:
No testcase defined.

lassign [:stateGet rollupCursor {0 -1}] fid off
for {set i 0} {$i<$batches} {incr i} {
    set rows [${:db} execute {
        SELECT fid, off FROM nsmcp_records.rows WHERE (fid, off)>(:fid, :off) AND rolled=0 ORDER BY fid, off
        LIMIT 512
    }]
    if {$rows eq {}} {
        :statePut rollupsReady 1
        return true
    }
    ${:db} execute {
        CREATE TEMP TABLE IF NOT EXISTS nsmcp_ingest_batch AS
        SELECT * FROM nsmcp_records.rows WHERE 0
    }
    ${:db} transaction {
        ${:db} execute {
            DELETE FROM temp.nsmcp_ingest_batch;
            INSERT INTO temp.nsmcp_ingest_batch SELECT * FROM nsmcp_records.rows WHERE (fid, off)>(:fid, :off)
            AND rolled=0 ORDER BY fid, off LIMIT 512
        }
        :rollBatch
        :observeBatch
        ${:db} execute {
            UPDATE nsmcp_records.rows SET rolled=1 WHERE (fid, off) IN(SELECT fid, off FROM
                                                                       temp.nsmcp_ingest_batch)
        }
        set last [lindex $rows end]
        set fid [dict get $last fid]
        set off [dict get $last off]
        :statePut rollupCursor [list $fid $off]
    }
}
return false

samples (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> samples selected \
    files redactor

Build deterministic samples from selected and successful requests. Restrict output to known source descriptors and apply the supplied redactor. Preserve truncation metadata when a stored user-agent label was clipped.

Parameters:

selected (required)
Trusted SQL predicate produced by the query-filter builder.

files (required)
Source descriptors used to attach sample locations.

redactor (required)
Redaction object used for sample output.

Returns:
A list of selected-request sample entries and successful-request sample entries.

Testcases:
No testcase defined.

set samples {}
set successes {}
foreach {field extra} {samples {} successSamples { AND status<400}} {
    set entries {}
    foreach row [${:db} execute "
        SELECT fid, off, t, status, path, method, pool, source, driver, auth, phase, label, clipped, hex(agent)
        AS agent, identifier, lower(hex(score)) AS score FROM temp.nsmcp_record_window WHERE $selected $extra
        ORDER BY score, fid, off LIMIT 5
    "] {
        set file {}
        foreach entry $files {
            if {[dict exists $entry archiveFid] && [dict get $entry archiveFid] eq [dict get $row fid]} {
                set file $entry
                break
            }
        }
        if {$file eq {}} {
            continue
        }
        set item [dict create  path [dict get $row path]  method [dict get $row method]  status [dict get $row status]  pool [dict get $row pool]  authenticationState [dict get $row auth]  startTime [dict get $row t]  recordSource [dict get $row source]  driver [dict get $row driver]  requestIdentifier [dict get $row identifier]  requestIdentifierUnique [expr {[dict get $row source] eq "connection"}]  phaseTimingAvailable [dict get $row phase]  phaseTimingReason [expr {
                                               [dict get $row phase] ? "logged-worker-phases" : ([dict get $row source] eq "driver" ?
                                                                                                 "driver-synthetic-phases" : "no-phase-counters")
                                           }]]
        if {[dict get $row agent] ne {}} {
            dict set item userAgent [dict get $row label]
        }
        set sample [accessScanSample $item $file [dict get $row off] $redactor]
        if {[dict get $row clipped]} {
            set index [lsearch -exact $sample userAgentTruncated]
            if {$index>=0} {
                lset sample [expr {$index+2}] true
            }
        }
        lappend entries [list [dict get $row score] $sample]
    }
    if {$field eq "samples"} {
        set samples $entries
    } else {
        set successes $entries
    }
}
return [list $samples $successes]

scope (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> scope files

Open a stored-record query scope for selected source intervals. Create connection-local range metadata and a bounded request view inside a read transaction. Reject a generation that disappeared after discovery. Call finish to end the scope.

Parameters:

files (required)
Source descriptors containing archive file identifiers and byte bounds.

Returns:
An empty result after the scoped transaction and view are ready.

Testcases:
No testcase defined.

set :fullScope false
if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=OFF}
}
${:db} execute {
    DROP VIEW IF EXISTS temp.nsmcp_record_window; DROP TABLE IF EXISTS temp.nsmcp_scan_batch; DROP TABLE IF
    EXISTS temp.nsmcp_record_scope;
    CREATE TEMP TABLE nsmcp_record_scope(fid INTEGER PRIMARY KEY,lo INTEGER,hi INTEGER)
}
foreach file $files {
    if {![dict exists $file archiveFid] || [dict get $file status] eq "outside-window"} {
        continue
    }
    set fid [dict get $file archiveFid]
    set lo [dict get $file archiveLo]
    set hi [dict get $file offset]
    ${:db} execute {INSERT INTO temp.nsmcp_record_scope VALUES(:fid,:lo,:hi)}
}
${:db} execute BEGIN
set :querying true
foreach file $files {
    if {[dict exists $file archiveFid]} {
        set fid [dict get $file archiveFid]
        if {$fid>=0 && ![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=:fid}]} {
            error "record-store-generation-changed"
        }
    }
}
${:db} execute {
    CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM nsmcp_records.rows r
    JOIN temp.nsmcp_record_scope s ON r.fid=s.fid AND r.off>=s.lo AND r.off<s.hi AND r.stop<=s.hi
}
if {${:mode} eq "reader"} {
    ${:db} execute {PRAGMA query_only=ON}
}

session (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> session
```
Return the database session owned by this adapter. Use the session for operations within the adapter's lifetime. The adapter retains responsibility for closing it and releasing the pool handle.

Returns:
The DatabaseSession object, or an empty value when no session is available.

Testcases:
No testcase defined.
```
return ${:db}
```
stateGet (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> stateGet key \
    [ default ]
```
Read a named control-state value. Return the stored Tcl representation without interpreting its contents. Use the supplied default when the key is absent.

Parameters:

key (required)
Control-state key.

default (optional)
Value to return when the key does not exist.

Returns:
The stored state value or the supplied default.

Testcases:
No testcase defined.
```
set rows [${:db} execute {SELECT value FROM nsmcp_records.state WHERE key=:key}]
return [expr {$rows eq {} ? $default : [dict get [lindex $rows 0] value]}]
```
statePut (scripted, public)
```
 <instance of nsmcp::AccessRecordStoreSQLite> statePut key value
```
Save a named control-state value. Replace the existing value using scalar binding. The caller controls transaction boundaries when this state accompanies record or coverage changes.

Parameters:

key (required)
Control-state key.

value (required)
Tcl value to serialize into the state table.

Returns:
An empty result after the value has been saved.

Testcases:
No testcase defined.
```
${:db} execute {INSERT OR REPLACE INTO nsmcp_records.state VALUES(:key,:value)}
```

stats (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> stats options \
    redactor limits

Aggregate request statistics for the current stored-record scope. Combine detailed request rows with eligible minute rollups. Bound ranked dimensions, redact returned labels and report omitted or unavailable observations explicitly.

Parameters:

options (required)
Normalized time window, filters, bucket size and result limits.

redactor (required)
Redaction object for returned dimension labels.

limits (required)
Aggregation limits, including the maximum number of retained keys.

Returns:
A list containing request statistics, aggregation metadata and the selected SQL predicate.

Testcases:
No testcase defined.

lassign [:filters $options] context selected
set from [dict get $options from]
set to [dict get $options to]
set stats [accessScanEmpty]
set rollup [:rollupSource $options]
if {$rollup ne {}} {
    set row [lindex [${:db} execute "
        SELECT coalesce(sum(n), 0) AS contextRecords, coalesce(sum(CASE WHEN status>=400 THEN n ELSE 0 END), 0)
        AS contextFailedRecords FROM ($rollup) WHERE $context
    "] 0]
} else {
    set row [lindex [${:db} execute "
        SELECT count(*) AS contextRecords, coalesce(sum(status>=400), 0) AS contextFailedRecords FROM
        temp.nsmcp_record_window WHERE $context
    "] 0]
}
set stats [dict merge $stats $row]
set row [lindex [${:db} execute "
    SELECT count(*) AS requestCount, coalesce(sum(status<400), 0) AS statusBelow400Records, coalesce(sum(bytes),
    0) AS responseBytes,
    coalesce(sum(phase), 0) AS timingCount, coalesce(sum(fallback), 0) AS fallbackCount, coalesce(sum(a), 0) AS
    acceptTotal, coalesce(sum(q), 0) AS queueTotal,
    coalesce(sum(f), 0) AS filterTotal, coalesce(sum(r), 0) AS runTotal, coalesce(max(r), 0) AS runMax,
    coalesce(min(t), '') AS matchedFirstAt, coalesce(max(t), '') AS matchedLastAt,
    coalesce(sum(peer=''), 0) AS unknownPeerRecords, coalesce(sum(agent=''), 0) AS unknownUserAgentRecords,
    coalesce(sum(clipped), 0) AS truncatedUserAgentRecords, coalesce(sum(redacted), 0) AS labelRedactedRecords,
    coalesce(sum(source='driver' AND status=503), 0) AS driver503Records, coalesce(sum(source='driver'), 0) AS
    driverRecords FROM temp.nsmcp_record_window WHERE $selected
"] 0]
set stats [dict merge $stats $row]
set row [lindex [${:db} execute {
    SELECT count(*) AS examinedRecords, coalesce(sum(kind=1), 0) AS parseErrors, coalesce(sum(kind=2), 0) AS
    oversizedRecords, coalesce(min(t), '') AS earliestAt, coalesce(max(t), '') AS latestAt FROM
    temp.nsmcp_record_window
}] 0]
set stats [dict merge $stats $row]
set counters [:checkpointCounts]
foreach key {parseErrors oversizedRecords} {
    dict incr stats $key [dict get $counters $key]
}
dict incr stats examinedRecords [dict get $counters checkpointRecords]
dict set stats diagnosticCountsComplete [dict get $counters diagnosticCountsComplete]
dict set stats pendingRecords [:boundaryPending]
foreach {field column cap} [list paths path [dict get $limits keys] methods method 32 statuses status 500 pools pool 64  hours {CAST(t/3600 AS INTEGER)*3600} 170 drivers driver 64] {
    set where $selected
    if {$field eq "drivers"} {
        append where " AND source='driver'"
    }
    set known 0
    if {$rollup ne {} && $field in {methods statuses pools hours}} {
        set sql "
            SELECT $column AS value, sum(n) AS n FROM ($rollup) WHERE $where GROUP BY $column ORDER BY n DESC,
            value LIMIT $cap
        "
    } else {
        set sql "
            SELECT $column AS value, count(*) AS n FROM temp.nsmcp_record_window WHERE $where GROUP BY $column
            ORDER BY n DESC, value LIMIT $cap
        "
    }
    foreach row [${:db} execute $sql] {
        set value [dict get $row value]
        set n [dict get $row n]
        if {$field ni {hours statuses}} {
            set value [$redactor message $value]
        }
        if {$field eq "paths" && [string length [encoding convertto utf-8 $value]]>1024} {
            continue
        }
        set counts [dict get $stats $field]
        dict incr counts $value $n
        dict set stats $field $counts
        incr known $n
    }
    set total  [dict get $stats [expr {$field eq "drivers" ? "driverRecords" : "requestCount"}]]
    if {$known<$total} {
        dict set stats omittedRecords $field [expr {$total-$known}]
    }
}
foreach {field column} {authenticationStates auth recordSources source} {
    if {$rollup ne {} && $field eq "recordSources"} {
        set sql  "SELECT $column AS value,sum(n) AS n FROM ($rollup) WHERE $selected GROUP BY $column"
    } else {
        set sql "
            SELECT $column AS value,count(*) AS n FROM temp.nsmcp_record_window WHERE $selected GROUP BY $column
        "
    }
    foreach row [${:db} execute $sql] {
        dict set stats $field [dict get $row value] [dict get $row n]
    }
}
foreach row [${:db} execute "
    SELECT status, count(*) AS requestCount, sum(phase) AS timingCount, sum(r) AS runTotal, max(r) AS runMax
    FROM temp.nsmcp_record_window WHERE $selected GROUP BY status
"] {
    set status [dict get $row status]
    dict unset row status
    dict set stats statusTimings $status $row
}
set seconds [dict get $options bucketSeconds]
set index "CAST((t-[ns_dbquotevalue $from])/$seconds AS INTEGER)"
foreach row [${:db} execute "
    SELECT $index AS i, count(*) AS n, sum(status>=400) AS failed, sum(phase) AS timings, sum(r) AS runtime FROM
    temp.nsmcp_record_window WHERE $context GROUP BY i LIMIT 3600
"] {
    dict set stats buckets [dict get $row i] [list contextRecords [dict get $row n] contextFailedRecords [dict get $row failed]  selectedRecords 0 driver503Records 0 timingCount 0 runTotal 0.0 contextTimingCount  [dict get $row timings] contextRunTotal [dict get $row runtime]]
}
foreach row [${:db} execute "
    SELECT $index AS i, count(*) AS n, sum(source='driver' AND status=503) AS driver503, sum(phase) AS timings,
    sum(r) AS runtime FROM temp.nsmcp_record_window WHERE $selected GROUP BY i LIMIT 3600
"] {
    set i [dict get $row i]
    foreach {key col}  {selectedRecords n driver503Records driver503 timingCount timings runTotal runtime} {
        dict set stats buckets $i $key [dict get $row $col]
    }
}
set aggregate [dict create  backend sqlite  cacheKiB 8192  status active  reason {}  maxBytes [dict get ${:settings} maxBytes]  diskBytes [:diskBytes]  peakBufferedKeys 512  peerOmitted 0  agentOmitted 0  labelTruncatedRecords 0  labelRedactedRecords 0]
foreach {field col} {peers peer agents agent} {
    set valid "$selected AND $col<>''"
    set totals [dict create  distinct 0  records 0  top1 0  top10 0  top100 0  rankedRecords 0  clippedRecords 0  redactedRecords 0]
    set entries {}
    set i 0
    foreach row [${:db} execute "
        SELECT *, count(*) OVER() AS distinctCount, sum(n) OVER() AS records FROM (SELECT hex($col) AS k,
        count(*) AS n, min(label) AS label, max(clipped) AS clipped, max(redacted) AS redacted FROM
        temp.nsmcp_record_window WHERE $valid GROUP BY $col) ORDER BY n DESC, k LIMIT 100
    "] {
        dict set totals distinct [dict get $row distinctCount]
        dict set totals records [dict get $row records]
        incr i
        set n [dict get $row n]
        if {$i==1} {
            dict set totals top1 $n
        }
        if {$i<=10} {
            dict incr totals top10 $n
        }
        dict incr totals top100 $n
        if {$i<=[dict get $options maxGroups]} {
            dict incr totals rankedRecords $n
            lappend entries [dict create  count $n  value [dict get $row label]  valueTruncated [dict get $row clipped]  valueRedacted [dict get $row redacted]]
        }
    }
    dict set aggregate $field $entries
    dict set aggregate ${field}Totals $totals
}
dict set aggregate labelTruncatedRecords [dict get $stats truncatedUserAgentRecords]
dict set aggregate labelRedactedRecords [dict get $stats labelRedactedRecords]
return [list $stats $aggregate $selected]

totals (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> totals options

Return stored request totals for the current query scope. Use minute rollups when the scope and filters permit, retaining detailed boundary rows. Otherwise aggregate the scoped records and checkpoint diagnostic counters directly.

Parameters:

options (required)
Normalized time window and request filters.

Returns:
A list containing the counts dictionary, selected status counts and coverage diagnostics.

Testcases:
No testcase defined.

lassign [:filters $options] context selected
set source [:rollupSource $options]
if {$source ne {}} {
    set counts [dict create requestCount 0 failedRecords 0 selectedRecords 0]
    set statuses {}
    # One grouped pass supplies both context and selected totals.
    foreach row [${:db} execute "
        SELECT status, sum(n) AS n, sum(CASE WHEN $selected THEN n ELSE 0 END) AS selected FROM ($source) WHERE
        $context GROUP BY status ORDER BY status
    "] {
        set status [dict get $row status]
        set n [dict get $row n]
        set selectedCount [dict get $row selected]
        dict incr counts requestCount $n
        if {$status>=400} {
            dict incr counts failedRecords $n
        }
        dict incr counts selectedRecords $selectedCount
        if {$selectedCount>0} {
            lappend statuses [dict create status $status count $selectedCount]
        }
    }
    set overview [:overview]
    return [list $counts $statuses [dict create  queryStrategy minute-rollups-with-boundaries  storedPhysicalRecords [dict get $counts requestCount]  parseErrors [dict get $overview malformed]  oversizedRecords [dict get $overview oversized]  pendingRecords 0]]
}
set counts [lindex [${:db} execute "
    SELECT count(*) AS requestCount, coalesce(sum(status>=400), 0) AS failedRecords, coalesce(sum(CASE WHEN
    $selected THEN 1 ELSE 0 END), 0) AS selectedRecords FROM temp.nsmcp_record_window WHERE $context
"] 0]
set statuses [${:db} execute "
    SELECT status, count(*) AS count FROM temp.nsmcp_record_window WHERE $selected GROUP BY status ORDER BY
    status
"]
set diagnostics [lindex [${:db} execute {
    SELECT count(*) AS storedPhysicalRecords, coalesce(sum(kind=1), 0) AS parseErrors, coalesce(sum(kind=2), 0)
    AS oversizedRecords FROM temp.nsmcp_record_window
}] 0]
set counters [:checkpointCounts]
foreach key {parseErrors oversizedRecords} {
    dict incr diagnostics $key [dict get $counters $key]
}
dict set diagnostics diagnosticCountsComplete [dict get $counters diagnosticCountsComplete]
dict set diagnostics queryStrategy detailed-rows
dict set diagnostics pendingRecords [:boundaryPending]
return [list $counts $statuses $diagnostics]

tuple (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> tuple fid off stop \
    kind row redactor file [ batched ]

Construct one policy-processed record from a valid parsed request. Redact display fields and encode the canonical row representation. Cache completion-time parsing and retain eligible peer-address observations for the enclosing ingestion transaction; do not insert a row here.

Parameters:

fid (required)
Stored source-file identifier.

off (required)
Starting source byte offset.

stop (required)
Exclusive byte offset after the record.

kind (required)
Record kind; only valid requests (zero) are accepted.

row (required)
Parsed request dictionary.

redactor (required)
Redaction object for display fields.

file (required)
Source descriptor used to derive the stable sample identifier.

batched (optional, defaults to "false")
True to return a prepared row; false to return a quoted SQL tuple.

Returns:
The cooked request as a Tcl row list or serialized SQL tuple.

Testcases:
No testcase defined.

if {$kind!=0} {
    error "only parsed requests have tuples"
}
set values [list $fid $off $stop $kind]
set peer [dict get $row peerAddress]
set agent [:dict getdef $row userAgent {}]
set peerBinary [expr  {$peer in {{} - {[REDACTED]}} ? {} : [ns_crypto::md string -digest sha256 -encoding binary -- $peer]}]
set peer [binary encode hex $peerBinary]
set key [expr {$agent in {{} -} ? {} : [ns_crypto::md string -digest sha256 -- $agent]}]
set label [$redactor message $agent]
set clipped [expr {
                   [string length [encoding convertto utf-8 $label]]>1024 ||
                   [string length [encoding convertto utf-8 $agent]]>1024
               }]
set redacted [expr {$label ne $agent}]
if {$clipped} {
    set label  [encoding convertfrom utf-8 [string range [encoding convertto utf-8 $label] 0 1023]]
    while {[string length [encoding convertto utf-8 $label]]>1024} {
        set label [string range $label 0 end-1]
    }
}
foreach k  {startTime status path method pool recordSource driver authenticationState responseBytes phaseTimingAvailable} {
    lappend values [dict get $row $k]
}
lappend values [expr {[dict get $row timingSource] eq "completion"}]
foreach k {acceptTime queueTime filterTime runTime} {
    lappend values [:dict getdef $row $k 0.0]
}
lappend values $peer $key $label $clipped $redacted [dict get $row requestIdentifier]  [ns_crypto::md string -digest sha256 -- [list [dict get $file identity] $off]]
set completed 0
if {$kind==0} {
    set stamp [dict get $row timestamp]
    if {$stamp ne ${:lastStamp}} {
        set :lastStamp $stamp
        if {[string is entier -strict $stamp]} {
            set :lastCompleted $stamp
        } elseif {[catch {
            clock scan $stamp -format {%d/%b/%Y:%H:%M:%S %z} -locale C
        } value]} {
            set :lastCompleted 0
        } else {
            set :lastCompleted $value
        }
    }
    set completed ${:lastCompleted}
    if {[:dict getdef ${:settings} unified false]} {
        set address [dict get $row peerAddress]
        set time [dict get $row startTime]
        if {
            [ns_ip valid $address] && $time>=[clock seconds]-[dict get ${:settings} peerRetention] &&
            $time<=[clock seconds]+60
        } {
            dict set :peers $peerBinary [list $address $time]
        }
    }
}
lappend values $completed 1
if {$batched} {
    return $values
}
# Hash columns retain BLOB storage; absent peer/agent hashes retain
# empty TEXT. Quote each contiguous scalar group in one core call.
set sql [ns_dbquotelist [lrange $values 0 18]]
foreach index {19 20} {
    set hash [lindex $values $index]
    if {$hash eq {}} {
        append sql ",''"
    } else {
        append sql ",X[ns_dbquotevalue $hash]"
    }
}
append sql , [ns_dbquotelist [lrange $values 21 24]]
append sql ,X [ns_dbquotevalue [lindex $values 25]]
append sql , [ns_dbquotelist [lrange $values 26 27]]
return "($sql)"

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables

Class ::nsmcp::AccessRecordStoreSQLite

Class Relations

Methods (to be applied on instances)

allScope (scripted, public)

available (scripted, public)

batchInsertionAvailable (scripted, public)

cRead (scripted, public)

capacity (scripted, public)

close (scripted, public)

coveredBytes (scripted, public)

coveredEnd (scripted, public)

descriptors (scripted, public)

diagnostic (scripted, public)

diskBytes (scripted, public)

dropState (scripted, public)

finish (scripted, public)

gapEnd (scripted, public)

ingest (scripted, public)

legacyViews (scripted, public)

observeBatch (scripted, public)

overview (scripted, public)

prepare (scripted, public)

prune (scripted, public)

rollupBackfill (scripted, public)

samples (scripted, public)

scope (scripted, public)

session (scripted, public)

stateGet (scripted, public)

statePut (scripted, public)

stats (scripted, public)

totals (scripted, public)

tuple (scripted, public)