# SPDX-License-Identifier: GPL-3.0-or-later
"""Storage backends for the AIS history recorder.

Two backends behind one interface:
  * GeoPackageStore  — a local, self-contained GeoPackage (SQLite). Built by hand
    (system tables + our tables, NO gpkg validation/RTree triggers) so the fast
    write path is plain ``sqlite3`` with WAL + batched executemany — OGR-created
    gpkgs carry triggers that call ST_* functions a vanilla sqlite3 connection
    lacks, which is why we do not use OGR here. QGIS reads the result normally.
  * PostgresStore    — the same logical schema on PostgreSQL/PostGIS, for the
    shared/multi-node deployment. ``needs_provisioning()`` lets the UI offer a
    one-click "create the tables" so the user never runs DDL by hand.

Three cumulative stages share the tables, joined by ``mmsi`` + ``voyage_id``
(UI names in parentheses):
  1 Textual   -> ``vessels`` (static) + ``events`` (coverage/nav/static changes)
  2 Voyages   -> ``voyages`` (LineString, stored only when no points are kept)
  3 Full      -> ``fixes``   (Point per fix, while a voyage is open)
"""
import re
import sqlite3
import struct
import time

SCHEMA_VERSION = 2      # v2: Postgres time columns are TIMESTAMPTZ (was REAL epoch)
_GPKG_APPLICATION_ID = 0x47504B47        # 'GPKG'


# --------------------------------------------------------------------------- #
#  GeoPackage geometry blob encoders (StandardGeoPackageBinary, LE, XY envelope)
# --------------------------------------------------------------------------- #
def _hdr(minx, maxx, miny, maxy, srs=4326):
    # magic 'GP', version 0, flags 0x03 (little-endian + XY envelope), srs, envelope
    return (b"GP" + bytes((0, 0x03)) + struct.pack("<i", srs)
            + struct.pack("<4d", minx, maxx, miny, maxy))


def point_blob(lon, lat, srs=4326):
    """gpkg blob for a POINT (lon, lat)."""
    return _hdr(lon, lon, lat, lat, srs) + struct.pack("<BI2d", 1, 1, lon, lat)


def line_blob(coords, srs=4326):
    """gpkg blob for a LINESTRING from [(lon, lat), ...] (>= 2 points)."""
    xs = [c[0] for c in coords]
    ys = [c[1] for c in coords]
    body = struct.pack("<BII", 1, 2, len(coords))
    body += b"".join(struct.pack("<2d", x, y) for x, y in coords)
    return _hdr(min(xs), max(xs), min(ys), max(ys), srs) + body


# --------------------------------------------------------------------------- #
#  Shared schema description (used by both backends)
# --------------------------------------------------------------------------- #
_ATTR_TABLES = ("history_nodes", "vessels", "events")   # aspatial
_FEATURE_TABLES = {"voyages": "LINESTRING", "fixes": "POINT"}


def _table_ddl(sql_geom):
    """Return {table: CREATE TABLE ...}. sql_geom(kind) yields the backend's
    geometry column clause (a gpkg BLOB, or a PostGIS geometry(...))."""
    return {
        "history_nodes": (
            "history_nodes (node_id TEXT PRIMARY KEY, created {ts}, "
            "plugin_version TEXT, logic_version TEXT, schema_version INTEGER, "
            "receiver_lat REAL, receiver_lon REAL)"),
        "vessels": (
            "vessels (mmsi TEXT PRIMARY KEY, name TEXT, callsign TEXT, imo TEXT, "
            "ship_type INTEGER, length REAL, beam REAL, draught REAL, "
            "first_seen {ts}, last_seen {ts}, node_id TEXT)"),
        "events": (
            "events (id INTEGER PRIMARY KEY {autoinc}, mmsi TEXT, voyage_id TEXT, "
            "t {ts}, kind TEXT, nav_status INTEGER, detail TEXT, node_id TEXT)"),
        "voyages": (
            "voyages (fid INTEGER PRIMARY KEY {autoinc}, {geom_line}, "
            "voyage_id TEXT UNIQUE, mmsi TEXT, node_id TEXT, start_t {ts}, "
            "end_t {ts}, end_reason TEXT, name TEXT, ship_type INTEGER, "
            "destination TEXT, eta TEXT, draught REAL)"),
        "fixes": (
            "fixes (fid INTEGER PRIMARY KEY {autoinc}, {geom_point}, "
            "voyage_id TEXT, mmsi TEXT, t {ts}, sog REAL, cog REAL, heading REAL, "
            "nav_status INTEGER, routeid INTEGER, route_km REAL, flow TEXT)"),
    }


class _BaseStore:
    """Common write API; subclasses provide connect/DDL/geometry encoding."""

    def ensure_indexes(self, stage):
        pass

    def close_dangling_voyages(self, node_id):
        """Finalize voyages left open by a previous session (they can never
        continue). Returns the count closed."""
        return 0

    def discard_voyage(self, voyage_id):
        """Delete a trivial / trackless voyage (and any of its fixes)."""
        pass

    def ensure_views(self, stage):
        pass

    def upsert_vessel(self, mmsi, static, node_id):
        raise NotImplementedError

    def add_event(self, ev, node_id):
        raise NotImplementedError

    def begin_voyage(self, voyage_id, mmsi, start_t, info, node_id):
        raise NotImplementedError

    def end_voyage(self, voyage_id, end_t, reason, line_coords=None):
        raise NotImplementedError

    def add_fix(self, fix):
        raise NotImplementedError

    def flush(self):
        raise NotImplementedError

    def close(self):
        raise NotImplementedError


# --------------------------------------------------------------------------- #
#  GeoPackage backend
# --------------------------------------------------------------------------- #
class GeoPackageStore(_BaseStore):
    def __init__(self, path):
        self.path = path
        self._con = sqlite3.connect(path)
        self._con.execute("PRAGMA application_id = %d" % _GPKG_APPLICATION_ID)
        self._con.execute("PRAGMA user_version = 10200")   # GeoPackage 1.2
        self._con.execute("PRAGMA journal_mode = WAL")
        self._con.execute("PRAGMA synchronous = NORMAL")
        self._fix_buf = []
        self._event_buf = []

    # ---- schema ---------------------------------------------------------- #
    def needs_provisioning(self):
        row = self._con.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='fixes'"
        ).fetchone()
        return not row or not row[0]

    def ensure_schema(self):
        if not self.needs_provisioning():
            return
        c = self._con
        c.executescript("""
        CREATE TABLE IF NOT EXISTS gpkg_spatial_ref_sys (
          srs_name TEXT NOT NULL, srs_id INTEGER NOT NULL PRIMARY KEY,
          organization TEXT NOT NULL, organization_coordsys_id INTEGER NOT NULL,
          definition TEXT NOT NULL, description TEXT);
        CREATE TABLE IF NOT EXISTS gpkg_contents (
          table_name TEXT NOT NULL PRIMARY KEY, data_type TEXT NOT NULL,
          identifier TEXT UNIQUE, description TEXT DEFAULT '',
          last_change DATETIME NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
          min_x DOUBLE, min_y DOUBLE, max_x DOUBLE, max_y DOUBLE, srs_id INTEGER);
        CREATE TABLE IF NOT EXISTS gpkg_geometry_columns (
          table_name TEXT NOT NULL, column_name TEXT NOT NULL,
          geometry_type_name TEXT NOT NULL, srs_id INTEGER NOT NULL,
          z TINYINT NOT NULL, m TINYINT NOT NULL,
          CONSTRAINT pk_geom_cols PRIMARY KEY (table_name, column_name));
        """)
        for srs_id, name, org, oid, defn in (
                (-1, "Undefined cartesian SRS", "NONE", -1, "undefined"),
                (0, "Undefined geographic SRS", "NONE", 0, "undefined"),
                (4326, "WGS 84", "EPSG", 4326, "GEOGCS[\"WGS 84\"]")):
            c.execute("INSERT OR IGNORE INTO gpkg_spatial_ref_sys VALUES "
                      "(?,?,?,?,?,?)", (name, srs_id, org, oid, defn, None))
        ddl = _table_ddl(None)
        geom = {"geom_line": "geom BLOB", "geom_point": "geom BLOB",
                "autoinc": "AUTOINCREMENT", "ts": "REAL"}   # SQLite: epoch REAL
        for t in _ATTR_TABLES + tuple(_FEATURE_TABLES):
            c.execute("CREATE TABLE IF NOT EXISTS " + ddl[t].format(**geom))
        for t in _ATTR_TABLES:
            c.execute("INSERT OR IGNORE INTO gpkg_contents (table_name, data_type,"
                      " identifier) VALUES (?, 'attributes', ?)", (t, t))
        for t, kind in _FEATURE_TABLES.items():
            c.execute("INSERT OR IGNORE INTO gpkg_contents (table_name, data_type,"
                      " identifier, srs_id) VALUES (?, 'features', ?, 4326)", (t, t))
            c.execute("INSERT OR IGNORE INTO gpkg_geometry_columns VALUES "
                      "(?, 'geom', ?, 4326, 0, 0)", (t, kind))
        c.commit()

    def ensure_indexes(self, stage):
        """Create the lookup indexes appropriate to the active stage."""
        c = self._con
        c.execute("CREATE INDEX IF NOT EXISTS events_mmsi_t ON events(mmsi, t)")
        if stage >= 2:
            c.execute("CREATE INDEX IF NOT EXISTS voyages_mmsi ON voyages(mmsi)")
        if stage >= 3:
            c.execute("CREATE INDEX IF NOT EXISTS fixes_voyage ON fixes(voyage_id)")
            c.execute("CREATE INDEX IF NOT EXISTS fixes_t ON fixes(t)")
            c.execute("CREATE INDEX IF NOT EXISTS fixes_mmsi_t ON fixes(mmsi, t)")
        c.commit()

    def ensure_views(self, stage):
        """A readable, joined browse view: one row per voyage with the vessel's
        identity and human UTC times (QGIS loads it as an attribute table).
        (A gpkg spatial view isn't cleanly renderable, so track geometry stays in
        the fixes/voyages tables + the 'Past voyages' retrieval.)"""
        self._con.executescript("""
        DROP VIEW IF EXISTS voyage_browse;
        CREATE VIEW voyage_browse AS
          SELECT v.voyage_id, v.mmsi, ve.name, ve.callsign, ve.ship_type,
                 datetime(v.start_t, 'unixepoch') AS start_utc,
                 datetime(v.end_t,   'unixepoch') AS end_utc,
                 v.end_reason, v.destination, v.draught, v.node_id
          FROM voyages v LEFT JOIN vessels ve ON ve.mmsi = v.mmsi;
        """)
        self._con.commit()

    def close_dangling_voyages(self, node_id):
        c = self._con
        # trackless orphans (no line, no fixes) -> discard; the rest -> close at
        # their last fix time, marked 'interrupted'.
        n = c.execute(
            "DELETE FROM voyages WHERE node_id=? AND end_t IS NULL AND geom IS NULL "
            "AND NOT EXISTS (SELECT 1 FROM fixes f WHERE f.voyage_id="
            "voyages.voyage_id)", (node_id,)).rowcount
        n += c.execute(
            "UPDATE voyages SET end_t=COALESCE((SELECT MAX(t) FROM fixes f WHERE "
            "f.voyage_id=voyages.voyage_id), start_t), end_reason='interrupted' "
            "WHERE node_id=? AND end_t IS NULL", (node_id,)).rowcount
        # finished but trackless (geom NULL, no fixes) -> legacy junk that shows
        # 'no stored track'; discard so only real tracks remain.
        n += c.execute(
            "DELETE FROM voyages WHERE node_id=? AND end_t IS NOT NULL AND "
            "geom IS NULL AND NOT EXISTS (SELECT 1 FROM fixes f WHERE "
            "f.voyage_id=voyages.voyage_id)", (node_id,)).rowcount
        c.commit()
        return n

    def discard_voyage(self, voyage_id):
        self._con.execute("DELETE FROM fixes WHERE voyage_id = ?", (voyage_id,))
        self._con.execute("DELETE FROM voyages WHERE voyage_id = ?", (voyage_id,))

    def register_node(self, node_id, plugin_version, logic_version,
                      receiver_lat=None, receiver_lon=None):
        self._con.execute(
            "INSERT OR REPLACE INTO history_nodes VALUES (?,?,?,?,?,?,?)",
            (node_id, time.time(), plugin_version, logic_version, SCHEMA_VERSION,
             receiver_lat, receiver_lon))
        self._con.commit()

    # ---- writes ---------------------------------------------------------- #
    def upsert_vessel(self, mmsi, s, node_id):
        self._con.execute(
            "INSERT INTO vessels (mmsi,name,callsign,imo,ship_type,length,beam,"
            "draught,first_seen,last_seen,node_id) VALUES (?,?,?,?,?,?,?,?,?,?,?) "
            "ON CONFLICT(mmsi) DO UPDATE SET name=COALESCE(excluded.name,name),"
            "callsign=COALESCE(excluded.callsign,callsign),"
            "imo=COALESCE(excluded.imo,imo),"
            "ship_type=COALESCE(excluded.ship_type,ship_type),"
            "length=COALESCE(excluded.length,length),"
            "beam=COALESCE(excluded.beam,beam),"
            "draught=COALESCE(excluded.draught,draught),"
            "last_seen=excluded.last_seen",
            (mmsi, s.get("name"), s.get("callsign"), s.get("imo"),
             s.get("ship_type"), s.get("length"), s.get("beam"), s.get("draught"),
             s.get("t"), s.get("t"), node_id))

    def add_event(self, ev, node_id):
        self._event_buf.append((ev.get("mmsi"), ev.get("voyage_id"), ev.get("t"),
                                ev.get("kind"), ev.get("nav_status"),
                                ev.get("detail"), node_id))

    def begin_voyage(self, voyage_id, mmsi, start_t, info, node_id):
        self._con.execute(
            "INSERT OR IGNORE INTO voyages (voyage_id,mmsi,node_id,start_t,name,"
            "ship_type,destination,eta,draught) VALUES (?,?,?,?,?,?,?,?,?)",
            (voyage_id, mmsi, node_id, start_t, info.get("name"),
             info.get("ship_type"), info.get("destination"), info.get("eta"),
             info.get("draught")))

    def end_voyage(self, voyage_id, end_t, reason, line_coords=None):
        geom = line_blob(line_coords) if line_coords and len(line_coords) >= 2 else None
        self._con.execute(
            "UPDATE voyages SET end_t=?, end_reason=?, geom=COALESCE(?,geom) "
            "WHERE voyage_id=?", (end_t, reason, geom, voyage_id))

    def add_fix(self, f):
        self._fix_buf.append((
            point_blob(f["lon"], f["lat"]), f.get("voyage_id"), f.get("mmsi"),
            f.get("t"), f.get("sog"), f.get("cog"), f.get("heading"),
            f.get("nav_status"), f.get("routeid"), f.get("route_km"), f.get("flow")))

    def flush(self):
        if self._event_buf:
            self._con.executemany(
                "INSERT INTO events (mmsi,voyage_id,t,kind,nav_status,detail,node_id)"
                " VALUES (?,?,?,?,?,?,?)", self._event_buf)
            self._event_buf.clear()
        if self._fix_buf:
            self._con.executemany(
                "INSERT INTO fixes (geom,voyage_id,mmsi,t,sog,cog,heading,"
                "nav_status,routeid,route_km,flow) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
                self._fix_buf)
            self._fix_buf.clear()
        self._con.commit()

    def close(self):
        try:
            self.flush()
            self._con.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        finally:
            self._con.close()


# --------------------------------------------------------------------------- #
#  PostgreSQL / PostGIS backend
# --------------------------------------------------------------------------- #
class PostgresStore(_BaseStore):
    """Same logical schema on PostGIS. Requires psycopg2 (import guarded). Needs a
    live server to exercise; the DDL + write SQL mirror the GeoPackage backend."""

    def __init__(self, conn, schema="ais_watchkeeper"):
        """conn is a libpq DSN string OR a dict of psycopg2 connect kwargs
        (host/port/dbname/user/password/...); credentials should already be
        resolved by the caller (never touch QgsAuthManager on the worker thread)."""
        import psycopg2                       # noqa: F401 (guarded dependency)
        # PostgreSQL identifiers cannot be bound as query parameters, so the schema
        # name is interpolated into the SQL below. Restrict it to a plain SQL
        # identifier here so that interpolation cannot inject (guards the B608s).
        if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", schema):
            raise ValueError("invalid PostgreSQL schema name: %r" % schema)
        self._pg = psycopg2
        self.schema = schema
        self._con = (psycopg2.connect(**conn) if isinstance(conn, dict)
                     else psycopg2.connect(conn))
        self._cur = self._con.cursor()
        self._fix_buf = []
        self._event_buf = []

    def needs_provisioning(self):
        self._cur.execute(
            "SELECT to_regclass(%s)", ("%s.fixes" % self.schema,))
        return self._cur.fetchone()[0] is None

    def ensure_schema(self):
        if not self.needs_provisioning():
            return
        s = self.schema
        self._cur.execute("CREATE SCHEMA IF NOT EXISTS %s" % s)
        self._cur.execute("CREATE EXTENSION IF NOT EXISTS postgis")
        ddl = _table_ddl(None)
        geom = {"geom_line": "geom geometry(LineString,4326)",
                "geom_point": "geom geometry(Point,4326)",
                "autoinc": "GENERATED BY DEFAULT AS IDENTITY",
                "ts": "TIMESTAMPTZ"}    # native time type -> QGIS temporal, BRIN,
        #                                 TimescaleDB-ready
        for t in _ATTR_TABLES + tuple(_FEATURE_TABLES):
            body = ddl[t].format(**geom)
            self._cur.execute("CREATE TABLE IF NOT EXISTS %s.%s" % (s, body))  # nosec B608
        self._con.commit()

    def ensure_indexes(self, stage):
        """Create the lookup indexes for the active stage. Stage 3 gets a BRIN
        index on the fix time column — tiny and ideal for the append-only,
        time-ordered ``fixes`` table (fast time-range scans without a fat btree),
        plus btrees for the per-voyage / per-vessel lookups the views need."""
        s = self.schema
        stmts = ["CREATE INDEX IF NOT EXISTS events_mmsi_t ON %s.events (mmsi, t)" % s]
        if stage >= 2:
            stmts.append("CREATE INDEX IF NOT EXISTS voyages_mmsi "
                         "ON %s.voyages (mmsi, start_t)" % s)
            stmts.append("CREATE INDEX IF NOT EXISTS voyages_geom_gist "
                         "ON %s.voyages USING gist (geom)" % s)   # area filter
        if stage >= 3:
            stmts.append("CREATE INDEX IF NOT EXISTS fixes_t_brin "
                         "ON %s.fixes USING brin (t)" % s)
            stmts.append("CREATE INDEX IF NOT EXISTS fixes_voyage "
                         "ON %s.fixes (voyage_id)" % s)
            stmts.append("CREATE INDEX IF NOT EXISTS fixes_mmsi_t "
                         "ON %s.fixes (mmsi, t)" % s)
            stmts.append("CREATE INDEX IF NOT EXISTS fixes_geom_gist "
                         "ON %s.fixes USING gist (geom)" % s)      # area filter
        for q in stmts:
            self._cur.execute(q)                                    # nosec B608
        self._con.commit()

    def ensure_views(self, stage):
        """A single renderable 'voyage_tracks' view: one line per voyage, joined to
        the vessel's identity, with geometry = the stored line or (stage 3) a line
        derived from the ordered fixes. QGIS loads it directly (unique id =
        voyage_id) — label it by name to browse every voyage on the map."""
        s = self.schema
        self._cur.execute(
            "CREATE OR REPLACE VIEW %s.voyage_tracks AS "  # nosec B608
            "SELECT v.voyage_id, v.mmsi, ve.name, ve.callsign, ve.ship_type, "
            "v.start_t, v.end_t, v.end_reason, v.destination, v.draught, v.node_id, "
            "COALESCE(v.geom, (SELECT ST_MakeLine(f.geom ORDER BY f.t) "
            "FROM %s.fixes f WHERE f.voyage_id = v.voyage_id)) AS geom "
            "FROM %s.voyages v LEFT JOIN %s.vessels ve ON ve.mmsi = v.mmsi"
            % (s, s, s, s))
        self._con.commit()

    def timescaledb_status(self):
        """(available, is_hypertable, message). Detection only — enabling the
        hypertable needs the extension installed server-side and a PK that
        includes the time column; kept as a documented manual follow-up."""
        try:
            self._cur.execute("SELECT 1 FROM pg_available_extensions "
                              "WHERE name='timescaledb'")
            if self._cur.fetchone() is None:
                return (False, False,
                        "timescaledb not installed on the server "
                        "(apt install timescaledb-2-postgresql-16 + "
                        "shared_preload_libraries)")
            self._cur.execute("SELECT 1 FROM pg_extension WHERE extname='timescaledb'")
            enabled = self._cur.fetchone() is not None
            hyper = False
            if enabled:
                self._cur.execute(
                    "SELECT 1 FROM timescaledb_information.hypertables "
                    "WHERE hypertable_schema=%s AND hypertable_name='fixes'",
                    (self.schema,))
                hyper = self._cur.fetchone() is not None
            return (True, hyper,
                    "available; fixes is%s a hypertable" % ("" if hyper else " not"))
        except Exception as exc:                        # noqa: BLE001
            return (False, False, str(exc))

    def close_dangling_voyages(self, node_id):
        s = self.schema
        self._cur.execute(
            "DELETE FROM %s.voyages v WHERE v.node_id=%%s AND v.end_t IS NULL AND "  # nosec B608
            "v.geom IS NULL AND NOT EXISTS (SELECT 1 FROM %s.fixes f WHERE "
            "f.voyage_id=v.voyage_id)" % (s, s), (node_id,))
        n = self._cur.rowcount
        self._cur.execute(
            "UPDATE %s.voyages AS v SET end_t=COALESCE((SELECT MAX(t) FROM %s.fixes "
            "f WHERE f.voyage_id=v.voyage_id), v.start_t), end_reason='interrupted' "  # nosec B608
            "WHERE v.node_id=%%s AND v.end_t IS NULL" % (s, s), (node_id,))
        n += self._cur.rowcount
        # finished but trackless (geom NULL, no fixes) -> legacy junk that shows
        # 'no stored track'; discard so only real tracks remain.
        self._cur.execute(
            "DELETE FROM %s.voyages v WHERE v.node_id=%%s AND v.end_t IS NOT NULL "
            "AND v.geom IS NULL AND NOT EXISTS (SELECT 1 FROM %s.fixes f WHERE "  # nosec B608
            "f.voyage_id=v.voyage_id)" % (s, s), (node_id,))
        n += self._cur.rowcount
        self._con.commit()
        return n

    def discard_voyage(self, voyage_id):
        self._cur.execute(
            "DELETE FROM %s.fixes WHERE voyage_id = %%s" % self.schema, (voyage_id,))  # nosec B608
        self._cur.execute(
            "DELETE FROM %s.voyages WHERE voyage_id = %%s" % self.schema,  # nosec B608
            (voyage_id,))

    def register_node(self, node_id, plugin_version, logic_version,
                      receiver_lat=None, receiver_lon=None):
        self._cur.execute(
            "INSERT INTO %s.history_nodes VALUES "  # nosec B608
            "(%%s,to_timestamp(%%s),%%s,%%s,%%s,%%s,%%s) "
            "ON CONFLICT (node_id) DO NOTHING" % self.schema,
            (node_id, time.time(), plugin_version, logic_version, SCHEMA_VERSION,
             receiver_lat, receiver_lon))
        self._con.commit()

    def upsert_vessel(self, mmsi, s, node_id):
        self._cur.execute(
            "INSERT INTO %s.vessels (mmsi,name,callsign,imo,ship_type,length,beam,"  # nosec B608
            "draught,first_seen,last_seen,node_id) VALUES "
            "(%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s,to_timestamp(%%s),to_timestamp(%%s),%%s) "
            "ON CONFLICT (mmsi) DO UPDATE SET last_seen=EXCLUDED.last_seen, "
            "name=COALESCE(EXCLUDED.name,%s.vessels.name)" % (self.schema, self.schema),
            (mmsi, s.get("name"), s.get("callsign"), s.get("imo"), s.get("ship_type"),
             s.get("length"), s.get("beam"), s.get("draught"), s.get("t"),
             s.get("t"), node_id))

    def add_event(self, ev, node_id):
        self._event_buf.append((ev.get("mmsi"), ev.get("voyage_id"), ev.get("t"),
                                ev.get("kind"), ev.get("nav_status"),
                                ev.get("detail"), node_id))

    def begin_voyage(self, voyage_id, mmsi, start_t, info, node_id):
        self._cur.execute(
            "INSERT INTO %s.voyages (voyage_id,mmsi,node_id,start_t,name,ship_type,"  # nosec B608
            "destination,eta,draught) VALUES "
            "(%%s,%%s,%%s,to_timestamp(%%s),%%s,%%s,%%s,%%s,%%s) "
            "ON CONFLICT (voyage_id) DO NOTHING" % self.schema,
            (voyage_id, mmsi, node_id, start_t, info.get("name"),
             info.get("ship_type"), info.get("destination"), info.get("eta"),
             info.get("draught")))

    def end_voyage(self, voyage_id, end_t, reason, line_coords=None):
        wkt = None
        if line_coords and len(line_coords) >= 2:
            wkt = "LINESTRING(%s)" % ",".join("%f %f" % (x, y) for x, y in line_coords)
        self._cur.execute(
            "UPDATE %s.voyages SET end_t=to_timestamp(%%s), end_reason=%%s, "  # nosec B608
            "geom=COALESCE(ST_GeomFromText(%%s,4326),geom) WHERE voyage_id=%%s"
            % self.schema, (end_t, reason, wkt, voyage_id))

    def add_fix(self, f):
        self._fix_buf.append((f["lon"], f["lat"], f.get("voyage_id"), f.get("mmsi"),
                              f.get("t"), f.get("sog"), f.get("cog"), f.get("heading"),
                              f.get("nav_status"), f.get("routeid"),
                              f.get("route_km"), f.get("flow")))

    def flush(self):
        if self._event_buf:
            self._cur.executemany(
                "INSERT INTO %s.events (mmsi,voyage_id,t,kind,nav_status,detail,"  # nosec B608
                "node_id) VALUES (%%s,%%s,to_timestamp(%%s),%%s,%%s,%%s,%%s)"
                % self.schema,
                self._event_buf)
            self._event_buf.clear()
        if self._fix_buf:
            self._cur.executemany(
                "INSERT INTO %s.fixes (geom,voyage_id,mmsi,t,sog,cog,heading,"  # nosec B608
                "nav_status,routeid,route_km,flow) VALUES "
                "(ST_SetSRID(ST_MakePoint(%%s,%%s),4326),%%s,%%s,to_timestamp(%%s),"
                "%%s,%%s,%%s,%%s,%%s,%%s,%%s)" % self.schema, self._fix_buf)
            self._fix_buf.clear()
        self._con.commit()

    def close(self):
        try:
            self.flush()
        finally:
            self._con.close()
