Receiver setup

// A minimal receiver: JSON and CSV batches in, one SQLite row per sample out.
// Node 22.13 or later, no dependencies. Run: node server.mjs
//
//   PORT          port to listen on (default 8080; 0 = any free port)
//   HOST          address to bind (default 127.0.0.1; use 0.0.0.0 to accept the phone)
//   DB_PATH       SQLite file (default samples.db)
//   RECEIVER_TOKEN  if set, require the header  Authorization: Bearer <token>
//
// POST any path: store the batch.  GET /stats: counts, as JSON.
import { createServer } from "node:http";
import { DatabaseSync } from "node:sqlite";

const PORT = Number(process.env.PORT ?? 8080);
const HOST = process.env.HOST ?? "127.0.0.1";
const TOKEN = process.env.RECEIVER_TOKEN ?? "";
const MAX_BODY = 2 * 1024 * 1024; // the app keeps every batch under 1 MB

const db = new DatabaseSync(process.env.DB_PATH ?? "samples.db");
db.exec(`
  CREATE TABLE IF NOT EXISTS samples (
    uuid TEXT PRIMARY KEY,          -- the deduplication key, stored exactly as sent
    id TEXT NOT NULL,               -- content fingerprint: repeats by design, never a key
    type TEXT NOT NULL, grp TEXT NOT NULL, source TEXT NOT NULL,
    start TEXT NOT NULL, "end" TEXT NOT NULL, kind TEXT NOT NULL,
    payload TEXT NOT NULL           -- the whole sample object, so added keys are kept
  );
  CREATE INDEX IF NOT EXISTS samples_id ON samples (id);
  CREATE TABLE IF NOT EXISTS deliveries (   -- one row per accepted request, retries included
    export_id TEXT NOT NULL, batch_sequence INTEGER NOT NULL, is_final INTEGER NOT NULL,
    basis TEXT, window_start TEXT, window_end TEXT, occasion_start TEXT, occasion_end TEXT,
    generated_at TEXT, format TEXT NOT NULL,
    samples INTEGER NOT NULL, inserted INTEGER NOT NULL, duplicates INTEGER NOT NULL,
    received_at TEXT NOT NULL
  );
`);
const exists = db.prepare("SELECT 1 FROM samples WHERE uuid = ?");
const upsert = db.prepare(`
  INSERT INTO samples (uuid, id, type, grp, source, start, "end", kind, payload)
  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT (uuid) DO UPDATE SET id = excluded.id, type = excluded.type, grp = excluded.grp,
    source = excluded.source, start = excluded.start, "end" = excluded."end",
    kind = excluded.kind, payload = excluded.payload`);
const logDelivery = db.prepare("INSERT INTO deliveries VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");

class Rejected extends Error {}

// Store one batch in one transaction: all of it or none of it, so a failed request can
// be retried by the app without leaving half a batch behind.
function store(envelope, samples, format) {
  let inserted = 0;
  let duplicates = 0;
  db.exec("BEGIN");
  try {
    for (const s of samples) {
      if (exists.get(s.uuid)) duplicates++;
      else inserted++;
      upsert.run(s.uuid, s.id, s.type, s.group, s.source, s.start, s.end, s.kind, JSON.stringify(s));
    }
    const e = envelope;
    logDelivery.run(e.exportId, e.batchSequence, e.isFinalBatch ? 1 : 0, e.basis ?? null,
      e.windowStart, e.windowEnd, e.occasionStart ?? null, e.occasionEnd ?? null, e.generatedAt,
      format, samples.length, inserted, duplicates, new Date().toISOString());
    db.exec("COMMIT");
  } catch (err) {
    db.exec("ROLLBACK");
    throw err;
  }
  return { inserted, duplicates };
}

function receiveJson(text) {
  let body;
  try {
    body = JSON.parse(text);
  } catch {
    throw new Rejected("body is not JSON");
  }
  if (body.schemaVersion !== 1) throw new Rejected(`unknown schemaVersion ${body.schemaVersion}`);
  if (body.connectivityTest) return { probe: true }; // acknowledge, record nothing
  return store(body, body.samples, "json");
}

// RFC 4180 parser: quoted fields may hold commas, doubled quotes and newlines.
function parseCsv(text) {
  const rows = [];
  let row = [];
  let field = "";
  let quoted = false;
  for (let i = 0; i < text.length; i++) {
    const c = text[i];
    if (quoted) {
      if (c === '"' && text[i + 1] === '"') { field += '"'; i++; }
      else if (c === '"') quoted = false;
      else field += c;
    } else if (c === '"') quoted = true;
    else if (c === ",") { row.push(field); field = ""; }
    else if (c === "\n") { row.push(field); rows.push(row); row = []; field = ""; }
    else if (c !== "\r") field += c;
  }
  if (field !== "" || row.length) { row.push(field); rows.push(row); }
  return rows;
}

// A CSV row becomes the same sample object the JSON rendering carries: numbers typed by
// `kind`, and an empty cell is an absent key, never 0.
function csvSample(r) {
  const s = { uuid: r.uuid, id: r.id, type: r.type, group: r.group, source: r.source,
    start: r.start, end: r.end, kind: r.kind };
  const num = (k) => { if (r[k] !== "") s[k] = Number(r[k]); };
  const str = (k) => { if (r[k] !== "") s[k] = r[k]; };
  if (r.kind === "quantity") { num("value"); str("unit"); }
  else if (r.kind === "category") { s.value = parseInt(r.value, 10); str("name"); }
  else if (r.kind === "workout") {
    s.activityType = parseInt(r.activityType, 10);
    str("activityName"); num("durationSeconds"); num("distanceMeters"); num("activeEnergyKilocalories");
  }
  return s;
}

function receiveCsv(text) {
  const [header, ...data] = parseCsv(text);
  if (!header || !header.includes("uuid")) throw new Rejected("no CSV header row");
  if (data.length === 0) return { probe: true }; // header, no rows: the probe
  // Parse by header name: columns may be appended, never reordered.
  const rows = data.map((cells) => Object.fromEntries(header.map((h, i) => [h, cells[i] ?? ""])));
  if (rows.some((r) => r.schemaVersion !== "1")) throw new Rejected("unknown schemaVersion");
  const r0 = rows[0];
  const envelope = { ...r0, batchSequence: Number(r0.batchSequence), isFinalBatch: r0.isFinalBatch === "true" };
  return store(envelope, rows.map(csvSample), "csv");
}

function stats() {
  const one = (sql) => Object.values(db.prepare(sql).get())[0] ?? 0;
  return {
    samples: one("SELECT COUNT(*) FROM samples"),
    duplicates: one("SELECT COALESCE(SUM(duplicates), 0) FROM deliveries"),
    deliveries: one("SELECT COUNT(*) FROM deliveries"),
    batches: one("SELECT COUNT(*) FROM (SELECT DISTINCT export_id, batch_sequence FROM deliveries)"),
  };
}

const server = createServer((req, res) => {
  const reply = (status, obj) => {
    res.writeHead(status, { "content-type": "application/json" }).end(JSON.stringify(obj));
  };
  if (req.method === "GET" && req.url === "/stats") return reply(200, stats());
  if (TOKEN && req.headers.authorization !== `Bearer ${TOKEN}`) return reply(401, { error: "credential" });
  if (req.method !== "POST") return reply(405, { error: "POST a batch" });

  const chunks = [];
  let size = 0;
  req.on("data", (c) => { size += c.length; if (size <= MAX_BODY) chunks.push(c); });
  req.on("end", () => {
    // A bare 413 is permanent: the app does not retry it.
    if (size > MAX_BODY) return reply(413, { error: "body too large" });
    const text = Buffer.concat(chunks).toString("utf8");
    const type = (req.headers["content-type"] ?? "").toLowerCase();
    try {
      if (type.startsWith("application/json")) return reply(200, receiveJson(text));
      if (type.startsWith("text/csv")) return reply(200, receiveCsv(text));
      return reply(415, { error: "expected application/json or text/csv" });
    } catch (err) {
      // A 4xx is not retried: only for a body that will never be accepted. Anything else
      // is our fault, and a 500 makes the app keep the batch and try again later.
      if (err instanceof Rejected) return reply(400, { error: err.message });
      console.error(err);
      return reply(500, { error: "internal" });
    }
  });
});

server.listen(PORT, HOST, () => console.log(`listening on ${HOST}:${server.address().port}`));

"""A minimal receiver: JSON and CSV batches in, one SQLite row per sample out.

Python 3.11 or later, standard library only. Run: python3 receiver.py

  PORT            port to listen on (default 8080; 0 = any free port)
  HOST            address to bind (default 127.0.0.1; use 0.0.0.0 to accept the phone)
  DB_PATH         SQLite file (default samples.db)
  RECEIVER_TOKEN  if set, require the header  Authorization: Bearer <token>

POST any path: store the batch.  GET /stats: counts, as JSON.
"""
import csv
import io
import json
import os
import sqlite3
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer

PORT = int(os.environ.get("PORT", "8080"))
HOST = os.environ.get("HOST", "127.0.0.1")
TOKEN = os.environ.get("RECEIVER_TOKEN", "")
MAX_BODY = 2 * 1024 * 1024  # the app keeps every batch under 1 MB

db = sqlite3.connect(os.environ.get("DB_PATH", "samples.db"), isolation_level=None)
db.executescript("""
  CREATE TABLE IF NOT EXISTS samples (
    uuid TEXT PRIMARY KEY,          -- the deduplication key, stored exactly as sent
    id TEXT NOT NULL,               -- content fingerprint: repeats by design, never a key
    type TEXT NOT NULL, grp TEXT NOT NULL, source TEXT NOT NULL,
    start TEXT NOT NULL, "end" TEXT NOT NULL, kind TEXT NOT NULL,
    payload TEXT NOT NULL           -- the whole sample object, so added keys are kept
  );
  CREATE INDEX IF NOT EXISTS samples_id ON samples (id);
  CREATE TABLE IF NOT EXISTS deliveries (   -- one row per accepted request, retries included
    export_id TEXT NOT NULL, batch_sequence INTEGER NOT NULL, is_final INTEGER NOT NULL,
    basis TEXT, window_start TEXT, window_end TEXT, occasion_start TEXT, occasion_end TEXT,
    generated_at TEXT, format TEXT NOT NULL,
    samples INTEGER NOT NULL, inserted INTEGER NOT NULL, duplicates INTEGER NOT NULL,
    received_at TEXT NOT NULL
  );
""")

UPSERT = """
  INSERT INTO samples (uuid, id, type, grp, source, start, "end", kind, payload)
  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT (uuid) DO UPDATE SET id = excluded.id, type = excluded.type, grp = excluded.grp,
    source = excluded.source, start = excluded.start, "end" = excluded."end",
    kind = excluded.kind, payload = excluded.payload"""


class Rejected(Exception):
    pass


def store(envelope, samples, fmt):
    """Store one batch in one transaction: all of it or none of it, so a failed request
    can be retried by the app without leaving half a batch behind."""
    inserted = duplicates = 0
    db.execute("BEGIN")
    try:
        for s in samples:
            if db.execute("SELECT 1 FROM samples WHERE uuid = ?", (s["uuid"],)).fetchone():
                duplicates += 1
            else:
                inserted += 1
            db.execute(UPSERT, (s["uuid"], s["id"], s["type"], s["group"], s["source"],
                                s["start"], s["end"], s["kind"], json.dumps(s)))
        e = envelope
        db.execute("INSERT INTO deliveries VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (
            e["exportId"], e["batchSequence"], 1 if e["isFinalBatch"] else 0, e.get("basis"),
            e["windowStart"], e["windowEnd"], e.get("occasionStart"), e.get("occasionEnd"),
            e["generatedAt"], fmt, len(samples), inserted, duplicates,
            datetime.now(timezone.utc).isoformat()))
        db.execute("COMMIT")
    except BaseException:
        db.execute("ROLLBACK")
        raise
    return {"inserted": inserted, "duplicates": duplicates}


def receive_json(text):
    try:
        body = json.loads(text)
    except ValueError:
        raise Rejected("body is not JSON")
    if body.get("schemaVersion") != 1:
        raise Rejected(f"unknown schemaVersion {body.get('schemaVersion')}")
    if body.get("connectivityTest"):
        return {"probe": True}  # acknowledge, record nothing
    return store(body, body["samples"], "json")


def csv_sample(r):
    """A CSV row becomes the same sample object the JSON rendering carries: numbers typed
    by `kind`, and an empty cell is an absent key, never 0."""
    s = {k: r[k] for k in ("uuid", "id", "type", "group", "source", "start", "end", "kind")}

    def num(k):
        if r[k] != "":
            s[k] = float(r[k])

    def text(k):
        if r[k] != "":
            s[k] = r[k]

    if r["kind"] == "quantity":
        num("value"); text("unit")
    elif r["kind"] == "category":
        s["value"] = int(r["value"]); text("name")
    elif r["kind"] == "workout":
        s["activityType"] = int(r["activityType"])
        text("activityName"); num("durationSeconds"); num("distanceMeters"); num("activeEnergyKilocalories")
    return s


def receive_csv(text):
    # A real CSV parser, read by header name: columns may be appended, never reordered.
    reader = csv.DictReader(io.StringIO(text, newline=""), restval="")
    if not reader.fieldnames or "uuid" not in reader.fieldnames:
        raise Rejected("no CSV header row")
    rows = list(reader)
    if not rows:
        return {"probe": True}  # header, no rows: the probe
    if any(r["schemaVersion"] != "1" for r in rows):
        raise Rejected("unknown schemaVersion")
    r0 = rows[0]
    envelope = {**r0, "batchSequence": int(r0["batchSequence"]), "isFinalBatch": r0["isFinalBatch"] == "true"}
    return store(envelope, [csv_sample(r) for r in rows], "csv")


def stats():
    one = lambda sql: db.execute(sql).fetchone()[0]
    return {
        "samples": one("SELECT COUNT(*) FROM samples"),
        "duplicates": one("SELECT COALESCE(SUM(duplicates), 0) FROM deliveries"),
        "deliveries": one("SELECT COUNT(*) FROM deliveries"),
        "batches": one("SELECT COUNT(*) FROM (SELECT DISTINCT export_id, batch_sequence FROM deliveries)"),
    }


class Handler(BaseHTTPRequestHandler):
    def reply(self, status, obj):
        data = json.dumps(obj).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def authorised(self):
        return not TOKEN or self.headers.get("Authorization") == f"Bearer {TOKEN}"

    def do_GET(self):
        if self.path == "/stats":
            return self.reply(200, stats())
        if not self.authorised():
            return self.reply(401, {"error": "credential"})
        self.reply(405, {"error": "POST a batch"})

    def do_POST(self):
        if not self.authorised():
            return self.reply(401, {"error": "credential"})
        size = int(self.headers.get("Content-Length") or 0)
        if size > MAX_BODY:
            return self.reply(413, {"error": "body too large"})  # a bare 413 is not retried
        text = self.rfile.read(size).decode("utf-8")
        ctype = (self.headers.get("Content-Type") or "").lower()
        try:
            if ctype.startswith("application/json"):
                return self.reply(200, receive_json(text))
            if ctype.startswith("text/csv"):
                return self.reply(200, receive_csv(text))
            return self.reply(415, {"error": "expected application/json or text/csv"})
        except Rejected as err:
            # A 4xx is not retried: only for a body that will never be accepted.
            return self.reply(400, {"error": str(err)})
        except Exception:
            # Anything else is our fault; a 500 makes the app keep the batch and retry.
            self.log_error("internal error")
            return self.reply(500, {"error": "internal"})

    def log_message(self, fmt, *args):
        pass  # keep request lines out of the output; errors still go to stderr


if __name__ == "__main__":
    server = HTTPServer((HOST, PORT), Handler)
    print(f"listening on {HOST}:{server.server_address[1]}", flush=True)
    server.serve_forever()