shopdb-flask/plugins/endpoints/models/endpoint.py

"""Endpoint RMM telemetry models.

Two shapes, one per concern:

  EndpointState  The cached CURRENT snapshot for an asset - one row per asset,
                 overwritten in place on every agent report. Holds the fast-
                 moving metrics (uptime, idle, last boot, logged-in user) and
                 the latest inventory snapshots (services/processes, open ports,
                 interfaces, scheduled tasks) as serialized JSON. This is the
                 cached-current-value half of the ledger pattern: it moves in
                 the SAME commit as any EndpointEvent rows the report carries.

  EndpointEvent  Append-only record of a curated Windows Event Log entry. The
                 agent only forwards a fixed catalog of IDs (logon/logoff,
                 privilege use, service install, unexpected shutdown, DCOM,
                 crash) so this stays a readable security/stability history
                 rather than a raw event firehose.

  EndpointLogCollection
                 A plaintext log file collected from an endpoint on operator
                 demand. Kept out of EndpointCommand.resultjson because it is
                 bulky, needs its own per-asset retention cap, and has to be
                 addressable by id for the download route.

Both link to the core Asset via assetid (FK, ON DELETE CASCADE), so an endpoint
is the same physical thing as its Computer/Asset - no duplicate identity. The
agent reports a hostname; the plugin resolves it to the existing Computer.
"""

import json
import re
from datetime import datetime

import sqlalchemy as sa

from shopdb.api import db


def _utciso(value):
    """Naive-UTC datetime -> ISO string marked UTC, or None.

    Every datetime this plugin stores is naive UTC. Serializing without the Z
    leaves the receiver to guess, and JavaScript guesses browser-local.
    """
    return value.isoformat() + 'Z' if value else None


class EndpointState(db.Model):
    """Latest reported state for a single endpoint (one row per asset)."""

    __tablename__ = 'endpointstates'

    endpointstateid = db.Column(db.Integer, primary_key=True)

    # The asset this state belongs to. Unique: exactly one current snapshot per
    # endpoint. Resolved from the reported hostname against the computers plugin.
    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        unique=True,
        nullable=False,
        index=True,
    )

    # Read-only view of the owning asset. No cascade or backref: the asset side
    # must not gain a dependency on this plugin.
    asset = db.relationship('Asset', lazy='joined', viewonly=True)

    hostname = db.Column(db.String(100), nullable=True)
    loggedinuser = db.Column(db.String(100), nullable=True)
    agentversion = db.Column(db.String(40), nullable=True)
    # The interactive tray app versions independently of the agent service, so
    # it is reported and cached as its own field (AGENT-CONTRACT.md 2). Null
    # when the endpoint has no tray installed or an older agent that predates
    # the field.
    trayversion = db.Column(db.String(40), nullable=True)

    # Fast-moving metrics
    lastboottime = db.Column(db.DateTime, nullable=True)
    uptimeseconds = db.Column(db.BigInteger, nullable=True)
    idleseconds = db.Column(db.BigInteger, nullable=True)
    cpupercent = db.Column(db.Float, nullable=True)
    memorypercent = db.Column(db.Float, nullable=True)

    # Latest inventory snapshots, serialized JSON (MEDIUMTEXT). Kept as the
    # newest snapshot rather than one row per sample: the agent reports the full
    # list each cycle and history of these lists is not what operators ask for.
    servicesjson = db.Column(db.Text(length=16777215), nullable=True)
    portsjson = db.Column(db.Text(length=16777215), nullable=True)
    interfacesjson = db.Column(db.Text(length=16777215), nullable=True)
    scheduledtasksjson = db.Column(db.Text(length=16777215), nullable=True)

    # Hardware and OS inventory, same newest-snapshot rule as above. These
    # arrived on the wire well before this and were logged as unknown keys, so
    # the shapes are already proven against real endpoint traffic.
    disksjson = db.Column(db.Text(length=16777215), nullable=True)
    biosjson = db.Column(db.Text(length=16777215), nullable=True)
    memoryjson = db.Column(db.Text(length=16777215), nullable=True)
    processorjson = db.Column(db.Text(length=16777215), nullable=True)
    osinfojson = db.Column(db.Text(length=16777215), nullable=True)

    # Session-scoped data collected by a probe running as the signed-in user
    # (mapped drives, user-owned listeners). Null when nobody is signed in,
    # which is a normal unattended state rather than a fault.
    usercontextjson = db.Column(db.Text(length=16777215), nullable=True)

    # Windows domain or workgroup membership. A scalar, not JSON: it is a single
    # value the fleet view filters and groups on.
    domain = db.Column(db.String(255), nullable=True)

    # True when the endpoint is holding a reboot. Null = never reported, which
    # is deliberately distinct from a confident False.
    pendingreboot = db.Column(db.Boolean, nullable=True)

    # Per-drive health from the agent's two disk sources. JSON because a machine
    # has a variable number of drives and each carries its own verdict. Kept
    # separate from disksjson: that is capacity, this is whether the hardware is
    # dying. Null = never reported.
    diskhealthjson = db.Column(db.Text(length=16777215), nullable=True)

    # Which signal(s) drove pendingreboot. That column is a single bool OR-ed
    # from three sources and so cannot say WHY, and the difference matters: a
    # pendingfilerename means files are staged for replacement on next boot, so
    # the box is mid-servicing, while a lone servicing key can sit set for weeks
    # on a machine nobody rebooted. Empty list = checked, nothing pending.
    # Null = never reported or the check failed.
    rebootreasonsjson = db.Column(db.Text(length=16777215), nullable=True)

    # Fast startup configuration and what the last boot actually did. Kept
    # because uptime is not trustworthy under fast startup: a box can report
    # days of uptime after what the operator experienced as a shutdown, and a
    # "reboot" that was a hybrid resume does not clear the state a support call
    # is usually trying to clear.
    bootinfojson = db.Column(db.Text(length=16777215), nullable=True)

    # Defender health. Persisted but deliberately NOT surfaced anywhere yet:
    # this environment produces known false positives, and wiring it to a card
    # now would train operators to ignore that card. Stored so the history is
    # already accumulating when someone is ready to interpret it.
    defenderjson = db.Column(db.Text(length=16777215), nullable=True)

    # Pending Dell Command | Update items, produced by the shopdb-dellupdates
    # agent utility from the log the \ShopDB-DellCommandScan task writes. Only
    # the count drives the status card; the short list is stored so the panel
    # can show what is actually pending without a second round trip.
    #
    # A null column and a stored payload with status 'unknown' both mean "never
    # scanned", which stays distinct from a scan that found nothing pending. The
    # status card relies on that distinction - see _dell_updates_field.
    dellupdatesjson = db.Column(db.Text(length=16777215), nullable=True)

    # Warn-only reconciler output, refreshed on every agent check-in in the same
    # commit as servicesjson. Each entry names a policy/shipped-attached monitor
    # the endpoint is expected to be running but the latest snapshot does not
    # report as running, so an operator sees the drift and intervenes manually.
    # This is warn-only: nothing here is auto-remediated. Null = never computed.
    driftjson = db.Column(db.Text(length=16777215), nullable=True)

    # Last-known MTConnect agent run state, cached so the WebUI toggle can show
    # a real state between command polls. Set by the agent's mtconnect_status
    # result. Null = never reported (unknown).
    mtconnectrunning = db.Column(db.Boolean, nullable=True)
    mtconnectcheckedat = db.Column(db.DateTime, nullable=True)

    # When the agent last successfully reported. There is no heartbeat separate
    # from the report, so silence past the quiet-hours window is the only
    # evidence an endpoint has stopped checking in.
    lastreportedat = db.Column(db.DateTime, nullable=True, index=True)

    createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    def __repr__(self):
        return f"<EndpointState {self.hostname or self.assetid}>"

    def to_dict(self):
        return {
            'endpointstateid': self.endpointstateid,
            'assetid': self.assetid,
            'hostname': self.hostname,
            'loggedinuser': self.loggedinuser,
            'agentversion': self.agentversion,
            'trayversion': self.trayversion,
            'lastboottime': _utciso(self.lastboottime),
            'uptimeseconds': self.uptimeseconds,
            'idleseconds': self.idleseconds,
            'cpupercent': self.cpupercent,
            'memorypercent': self.memorypercent,
            'domain': self.domain,
            'pendingreboot': self.pendingreboot,
            'mtconnectrunning': self.mtconnectrunning,
            'mtconnectcheckedat': _utciso(self.mtconnectcheckedat),
            'lastreportedat': _utciso(self.lastreportedat),
            'createdat': _utciso(self.createdat),
        }


class EndpointCommand(db.Model):
    """A queued operator command targeting one endpoint agent.

    The WebUI enqueues a named action (server allowlist only, never an arbitrary
    command string) for an asset. The agent claims the next pending command on
    its long-poll, executes it locally, and reports the outcome back, which
    moves the row to a terminal status in the same request. This is a small
    at-most-one-in-flight queue, not a general job system.
    """

    __tablename__ = 'endpointcommands'

    # Lifecycle: pending -> running -> succeeded|failed. A command is claimed
    # (pending -> running) by exactly one agent poll, then finalized by its
    # result post. A command that is claimed but whose agent never reports back
    # (agent killed/uninstalled/offline mid-flight) would otherwise pin the
    # at-most-one guard forever; the enqueue path reaps such rows to STATUS_EXPIRED
    # so a dead in-flight command cannot wedge the endpoint's command queue.
    STATUS_PENDING = 'pending'
    STATUS_RUNNING = 'running'
    STATUS_SUCCEEDED = 'succeeded'
    STATUS_FAILED = 'failed'
    STATUS_EXPIRED = 'expired'

    # In-flight staleness ceilings. A RUNNING command is expected to finish in
    # seconds to a couple of minutes; the generous ceiling tolerates a slow
    # app install while still freeing the queue when the agent is gone. A
    # PENDING command is claimed within one poll (~25s) by a live agent, so a
    # long-unclaimed row means the agent is offline and the queue should not
    # stay wedged behind it.
    RUNNING_STALE_SECONDS = 15 * 60
    PENDING_STALE_SECONDS = 60 * 60

    endpointcommandid = db.Column(db.Integer, primary_key=True)

    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        nullable=False,
        index=True,
    )

    asset = db.relationship('Asset', lazy='joined', viewonly=True)

    # The allowlisted action name (for example 'mtconnect_start'). Validated
    # against the server allowlist at enqueue time, so only known names land.
    action = db.Column(db.String(64), nullable=False)

    # Generic actions (service_* / process_*) carry their subject over the wire:
    # target is a Windows service name or a process image (for example
    # 'DncMain.exe'); targetkind is 'service' or 'process'. Fixed-target actions
    # (mtconnect_*) leave both null. This is what lets a site build the agent
    # once and drive every future policy/app purely from server data.
    target = db.Column(db.String(255), nullable=True)
    targetkind = db.Column(db.String(16), nullable=True)

    status = db.Column(db.String(20), nullable=False,
                       default=STATUS_PENDING, index=True)

    # Who queued it (userid) for audit; the agent never sees this.
    requestedby = db.Column(db.Integer, nullable=True)

    # The content digest as it stood WHEN AN OPERATOR APPROVED THIS PUSH.
    #
    # Delivery descriptors are re-resolved at claim time, minutes or hours
    # later, so without this the bytes an operator approved and the bytes the
    # agent runs are two different lookups against mutable content. Whoever can
    # replace a script body or an installer package could swap it in the gap
    # and the digest would faithfully describe the new bytes.
    #
    # Set at enqueue for run_script and update_app; null for actions that
    # deliver no content. The claim path refuses a command whose descriptor no
    # longer matches, so a swap fails the command instead of shipping.
    pinnedsha256 = db.Column(db.String(64), nullable=True)

    # Agent-reported outcome. resultjson holds structured detail (for example
    # the observed MTConnect running state); resultmessage is a short summary.
    resultmessage = db.Column(db.Text, nullable=True)
    resultjson = db.Column(db.Text(length=16777215), nullable=True)

    createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    claimedat = db.Column(db.DateTime, nullable=True)
    completedat = db.Column(db.DateTime, nullable=True)

    __table_args__ = (
        db.Index('idx_endpointcommands_assetstatus', 'assetid', 'status'),
    )

    def __repr__(self):
        return f"<EndpointCommand {self.action} asset={self.assetid} {self.status}>"

    def to_dict(self, includedetail=False):
        """Serialise a command for the operator views.

        resultjson is opt-in. It is agent-reported structured detail, and for a
        script it carries the evidence that a green badge alone does not: the
        exit code, the context it ran as and the directory it really ran in. On
        failure the agent also puts the captured output in there, so it is not
        served to every list that only needs a status.
        """
        data = {
            'endpointcommandid': self.endpointcommandid,
            'assetid': self.assetid,
            'action': self.action,
            'target': self.target,
            'targetkind': self.targetkind,
            'status': self.status,
            'requestedby': self.requestedby,
            'resultmessage': self.resultmessage,
            'createdat': _utciso(self.createdat),
            'claimedat': _utciso(self.claimedat),
            'completedat': _utciso(self.completedat),
        }
        if includedetail:
            data['resultdetail'] = self._resultdetail()
        return data

    def _resultdetail(self):
        """resultjson as a dict, or None.

        Stored as text, so a malformed or truncated payload is possible. A
        detail panel is diagnostics: failing to parse it must not break the
        command list that surrounds it.
        """
        if not self.resultjson:
            return None
        try:
            parsed = json.loads(self.resultjson)
        except (TypeError, ValueError):
            return None
        return parsed if isinstance(parsed, dict) else None


class EndpointEvent(db.Model):
    """A single curated Windows Event Log record forwarded by the agent."""

    __tablename__ = 'endpointevents'

    endpointeventid = db.Column(db.Integer, primary_key=True)

    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        nullable=False,
        index=True,
    )

    asset = db.relationship('Asset', lazy='joined', viewonly=True)

    # The Windows Event ID (for example 4625 failed logon, 41 unexpected
    # shutdown, 1001 application crash). Indexed because alerting queries filter
    # on it.
    eventid = db.Column(db.Integer, nullable=False, index=True)

    # Event Log channel the record came from: Security, System, Application.
    eventchannel = db.Column(db.String(50), nullable=True)
    eventsource = db.Column(db.String(255), nullable=True)

    # The Windows event record id: monotonic and unique within its channel on a
    # given machine. This is the dedup key. The agent also uses it as its
    # harvest watermark, so in normal operation a record is sent once; the
    # unique index below is what guarantees it even when an agent is reinstalled
    # (losing its watermark) or two reports overlap. Nullable because an agent
    # older than this field still reports, and MySQL permits repeated NULLs in a
    # unique index, so those degrade to the previous append-only behavior rather
    # than failing.
    eventrecordid = db.Column(db.BigInteger, nullable=True)

    # When the event happened on the endpoint (from the record), not when it
    # reached ShopDB.
    eventtime = db.Column(db.DateTime, nullable=True, index=True)

    eventmessage = db.Column(db.Text, nullable=True)

    # Structured fields the agent extracted for this ID (for example the target
    # account on a 4625). Advisory index only, not part of dedup.
    payloadjson = db.Column(db.Text(length=16777215), nullable=True)

    recordedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    __table_args__ = (
        db.Index('idx_endpointevents_assetevent', 'assetid', 'eventtime'),
        db.Index('idx_endpointevents_assetrecord',
                 'assetid', 'eventchannel', 'eventrecordid', unique=True),
    )

    def __repr__(self):
        return f"<EndpointEvent {self.eventid} asset={self.assetid}>"

    def to_dict(self):
        return {
            'endpointeventid': self.endpointeventid,
            'assetid': self.assetid,
            'eventid': self.eventid,
            'eventrecordid': self.eventrecordid,
            'eventchannel': self.eventchannel,
            'eventsource': self.eventsource,
            'eventtime': _utciso(self.eventtime),
            'eventmessage': self.eventmessage,
            'recordedat': _utciso(self.recordedat),
        }


class EndpointApplication(db.Model):
    """One installed-software (SBOM) entry reported for an endpoint.

    The agent enumerates both Windows Uninstall registry hives and emits a
    per-item {displayname, displayversion, publisher}, de-duped on
    name+version. The server stores that as a per-asset SNAPSHOT: on each
    report the existing rows for the asset are replaced with the reported set,
    mirroring how the services/ports inventories are kept as the newest list
    rather than an append-only history. History of installs is not what
    operators ask for; the current inventory is.

    appid is groundwork for a future link to the core applications catalog
    (the "Managed Applications" concept): a nullable, unenforced-by-match FK
    that a later matcher can populate to associate a discovered install with a
    managed Application row. No matching logic exists yet - the column and the
    relationship are laid down here so the association can be added without a
    second migration.
    """

    __tablename__ = 'endpointapplications'

    # BigInteger on MySQL, Integer on SQLite. This table is rewritten wholesale
    # on every check-in, so the key never stops climbing and a 32-bit column
    # eventually stops accepting inserts - see the migration for the arithmetic.
    #
    # The variant is not decoration: SQLite only autoincrements a column typed
    # exactly INTEGER PRIMARY KEY, so a plain BigInteger makes every insert fail
    # with a NOT NULL violation on the test database. Nothing caught that,
    # because no test inserts into this table.
    endpointapplicationid = db.Column(
        db.BigInteger().with_variant(db.Integer, 'sqlite'), primary_key=True)

    # The asset this install belongs to. Not unique: many applications per
    # endpoint. CASCADE so removing the asset clears its SBOM snapshot.
    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        nullable=False,
        index=True,
    )

    asset = db.relationship('Asset', lazy='joined', viewonly=True)

    # Wire fields, stored under the same lowercase/no-underscore names the agent
    # emits. displayname is always present (the agent skips entries without it);
    # displayversion and publisher default to '' on the agent side.
    displayname = db.Column(db.String(255), nullable=False, index=True)
    displayversion = db.Column(db.String(128), nullable=True)
    publisher = db.Column(db.String(255), nullable=True)

    # Groundwork link to the core applications catalog ("Managed Applications").
    # Nullable and unpopulated for now: no discovery-to-catalog matcher is wired
    # yet. SET NULL so retiring a managed Application does not delete discovered
    # SBOM rows - the install is still on the machine, it is just no longer
    # managed.
    appid = db.Column(
        db.Integer,
        db.ForeignKey('applications.appid', ondelete='SET NULL'),
        nullable=True,
        index=True,
    )

    application = db.relationship('Application', lazy='select', viewonly=True)

    # When this snapshot row was written (the report that produced it).
    recordedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    # SHA256 fingerprint for delta-based updates: hash of displayname, displayversion,
    # publisher. Allows efficient comparison to detect changes without comparing all
    # columns. Computed by agent and sent in payload. Nullable for backward compatibility
    # during migration - backfill script will populate existing rows.
    appfingerprint = db.Column(db.String(64), nullable=True)

    __table_args__ = (
        db.Index('idx_endpointapplications_assetname', 'assetid', 'displayname'),
        db.Index('idx_endpointapplications_fingerprint', 'assetid', 'appfingerprint', unique=True),
    )

    def __repr__(self):
        return f"<EndpointApplication {self.displayname!r} asset={self.assetid}>"

    def to_dict(self):
        return {
            'endpointapplicationid': self.endpointapplicationid,
            'assetid': self.assetid,
            'displayname': self.displayname,
            'displayversion': self.displayversion,
            'publisher': self.publisher,
            'appid': self.appid,
            'recordedat': _utciso(self.recordedat),
        }


class EndpointLock(db.Model):
    """An admin-only quarantine refusing operator commands on one endpoint.

    Deliberately NOT the same thing as a PolicyActionHold, despite the similar
    name. The two sit on different axes and collapsing them would recreate the
    axis-mixing this refactor exists to undo:

      PolicyActionHold  policy-resolution axis. Suppresses ONE desired-state
                        action so reconciliation stops fighting an admin.
                        Per-action, several may stand at once, and it EXPIRES.
      EndpointLock      caller-authority axis. Refuses commands from non-admins
                        on this endpoint regardless of policy or permission.
                        Per-endpoint, one at a time, and INDEFINITE.

    Indefinite is the point. A lock represents a physical-world condition - a
    machine mid-repair, a validated part running, a box that must not reboot
    during a customer audit - and those do not end on a timer. A TTL here would
    quietly unlock a machine someone is still working inside, which is the exact
    accident the lock exists to prevent. Stale locks are surfaced in the UI
    instead of being auto-cleared.

    Released rows are kept and marked rather than deleted: the record of who
    quarantined a machine, why, and who lifted it is the audit trail, and it is
    most wanted precisely when the lock is no longer there to explain itself.
    """

    __tablename__ = 'endpointlocks'

    endpointlockid = db.Column(db.Integer, primary_key=True)

    # Not unique at the schema level: released rows accumulate per asset and a
    # unique constraint would make the second lock in an endpoint's life fail.
    # The one-active-lock invariant is enforced on write in PLACE_ENDPOINT_LOCK.
    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        nullable=False,
        index=True,
    )

    reason = db.Column(db.Text, nullable=False)

    # Who set it. The id goes null if the account is deleted, but the username
    # snapshot survives: a quarantine that forgets who imposed it is unusable at
    # review time, which is the only time anyone reads it.
    lockedbyuserid = db.Column(
        db.Integer,
        db.ForeignKey('users.userid', ondelete='SET NULL'),
        nullable=True,
    )
    lockedbyusername = db.Column(db.String(100), nullable=True)
    lockedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

    # Null while the lock stands. Set on release, which is what makes this row
    # inactive - there is no isactive flag to drift out of step with them.
    releasedat = db.Column(db.DateTime, nullable=True)
    releasedbyuserid = db.Column(
        db.Integer,
        db.ForeignKey('users.userid', ondelete='SET NULL'),
        nullable=True,
    )
    releasedbyusername = db.Column(db.String(100), nullable=True)

    __table_args__ = (
        # Serves the hot path: "is there an active lock on this asset", asked on
        # every enqueue and every detail page load.
        db.Index('idx_endpointlocks_assetactive', 'assetid', 'releasedat'),
    )

    def __repr__(self):
        state = 'released' if self.releasedat else 'active'
        return f"<EndpointLock asset={self.assetid} {state}>"

    def to_dict(self):
        return {
            'endpointlockid': self.endpointlockid,
            'assetid': self.assetid,
            'reason': self.reason,
            'lockedbyuserid': self.lockedbyuserid,
            'lockedbyusername': self.lockedbyusername,
            'lockedat': _utciso(self.lockedat),
            'releasedat': _utciso(self.releasedat),
            'releasedbyusername': self.releasedbyusername,
        }


class EndpointLogCollection(db.Model):
    """One plaintext log file collected from an endpoint by an operator.

    Collected content lives here rather than in EndpointCommand.resultjson for
    three reasons. It is bulky and would be dragged along by every command
    history list that only wants a status; it needs its own retention rule
    (RETENTION_PER_ASSET), which a command row cannot express without deleting
    the audit record of the command itself; and the download route needs to
    serve bytes, which means it needs a row it can address by id rather than a
    JSON field it has to re-parse.

    Retention is deliberately PER ASSET. A global cap would let one busy machine
    evict another machine's history, and the history is read per machine on the
    PC detail page, so a global cap would make the panel unreliable exactly
    where it is used. Pruning happens on insert in the result route.

    The content stored here is a TAIL, not necessarily the whole file - see
    istruncated. A log that has run for months is not something to load into a
    browser pane, and the agent bounds what it sends.
    """

    __tablename__ = 'endpointlogcollections'

    # How many collections to keep per asset. The operator asked for 20: enough
    # to compare a handful of runs against each other, few enough that stored
    # log text does not grow without bound across a fleet.
    RETENTION_PER_ASSET = 20

    endpointlogcollectionid = db.Column(db.Integer, primary_key=True)

    assetid = db.Column(
        db.Integer,
        db.ForeignKey('assets.assetid', ondelete='CASCADE'),
        nullable=False,
        index=True,
    )

    # The command that produced this, kept for audit: it ties the collected
    # bytes back to who queued the collection and when it ran. SET NULL rather
    # than CASCADE so pruning or expiring a command never silently destroys a
    # log an operator is still looking at.
    endpointcommandid = db.Column(
        db.Integer,
        db.ForeignKey('endpointcommands.endpointcommandid', ondelete='SET NULL'),
        nullable=True,
    )

    # The path as the AGENT resolved it, which is not always the path the
    # operator typed (environment variables get expanded endpoint-side). Storing
    # the resolved one is what makes the history honest about what was read.
    sourcepath = db.Column(db.String(512), nullable=False)

    content = db.Column(db.Text(length=16777215), nullable=True)

    # Size of the file ON DISK, not the size of content. When the tail is
    # truncated these differ, and that difference is the whole point: it tells
    # the operator they are looking at the end of something larger.
    filesizebytes = db.Column(db.BigInteger, nullable=True)
    contentbytes = db.Column(db.Integer, nullable=True)
    linecount = db.Column(db.Integer, nullable=True)
    istruncated = db.Column(db.Boolean, nullable=False,
                            default=False, server_default=sa.false())

    # Last-write time reported by the endpoint. A log that has not changed since
    # the previous collection is the fastest way to see an app is not logging.
    filemodifiedat = db.Column(db.DateTime, nullable=True)

    # Username snapshot alongside the id: the id alone becomes unreadable if the
    # user is later removed, and this history is read long after collection.
    collectedbyuserid = db.Column(
        db.Integer,
        db.ForeignKey('users.userid', ondelete='SET NULL'),
        nullable=True,
    )
    collectedbyusername = db.Column(db.String(100), nullable=True)
    collectedat = db.Column(db.DateTime, nullable=False,
                            default=datetime.utcnow, index=True)

    __table_args__ = (
        # Serves both hot paths: listing an asset's history newest-first, and
        # finding the oldest rows to prune on insert.
        db.Index('idx_endpointlogcollections_assetcollected',
                 'assetid', 'collectedat'),
    )

    def __repr__(self):
        return (f"<EndpointLogCollection asset={self.assetid} "
                f"{self.sourcepath}>")

    @property
    def filename(self):
        """A download filename derived from the collected path."""
        raw = (self.sourcepath or '').replace('/', '\\')
        base = raw.rsplit('\\', 1)[-1].strip() or 'collected.log'
        # Keep it to characters that survive a Content-Disposition header and a
        # Windows save dialog without quoting games.
        safe = re.sub(r'[^A-Za-z0-9._-]', '_', base)[:100]
        stamp = (self.collectedat or datetime.utcnow()).strftime('%Y%m%d-%H%M%S')
        return f'{stamp}_{safe}'

    def to_dict(self, includecontent=False):
        """Serialise a collection.

        Content is opt-in for the same reason resultjson is on EndpointCommand:
        the history list renders 20 rows and none of them need the log body,
        which may be hundreds of kilobytes each.
        """
        data = {
            'endpointlogcollectionid': self.endpointlogcollectionid,
            'assetid': self.assetid,
            'endpointcommandid': self.endpointcommandid,
            'sourcepath': self.sourcepath,
            'filesizebytes': self.filesizebytes,
            'contentbytes': self.contentbytes,
            'linecount': self.linecount,
            'istruncated': bool(self.istruncated),
            'filemodifiedat': _utciso(self.filemodifiedat),
            'collectedbyuserid': self.collectedbyuserid,
            'collectedbyusername': self.collectedbyusername,
            'collectedat': _utciso(self.collectedat),
            'filename': self.filename,
        }
        if includecontent:
            data['content'] = self.content or ''
        return data