smb-online/backoffice/app/app.py

"""smb-crm back-office service.

Postgres is the source of truth. This serves the operator dashboard + a small
JSON API (read now; add/edit/delete and the DB->Sheets mirror layered on next).
Browser access is gated by Caddy basic-auth on onboard.mivanchenko.de; the
machine-to-machine ingest path (n8n) is gated by the CRM_API_TOKEN header.
"""
import os
import re
import time
import queue
import threading
import traceback
from datetime import datetime, date, timezone
from decimal import Decimal

from flask import Flask, jsonify, request, Response
from waitress import serve

import db
from sheets import Sheets

app = Flask(__name__, static_folder="static", static_url_path="")

CRM_TOKEN = os.environ.get("CRM_API_TOKEN", "")
# Separate read-only token for the public iCal feed (calendar apps can't send
# the basic-auth header, so the feed is gated by this query token instead).
ICS_TOKEN = os.environ.get("ICS_TOKEN", "")

# DB -> Sheets one-way mirror. Postgres is the source of truth; the Sheet is a
# best-effort projection. A mirror failure never fails the DB write.
SH = None
try:
    SH = Sheets(os.environ["GOOGLE_SA_JSON"], os.environ["SHEET_ID"])
except Exception:  # noqa: BLE001
    traceback.print_exc()


def _cell(v):
    if v is None:
        return ""
    if isinstance(v, bool):
        return "TRUE" if v else "FALSE"
    if isinstance(v, datetime):
        return v.strftime("%Y-%m-%d %H:%M")
    if isinstance(v, date):
        return v.strftime("%Y-%m-%d")
    if isinstance(v, Decimal):
        f = float(v)
        return str(int(f)) if f == int(f) else str(f)
    s = str(v)
    return ("'" + s) if s[:1] in "=+-@" else s  # neutralise formula injection


def mirror_entity(entity):
    if SH is None:
        return 0
    spec = db.TABLES[entity]
    cols = spec["cols"]
    order = LIST_ORDER.get(entity, spec["pk"])
    with db.connect() as conn, conn.cursor() as cur:
        cur.execute(f"SELECT {', '.join(cols)} FROM {entity} ORDER BY {order}")
        rows = cur.fetchall()
    grid = [[_cell(r[c]) for c in cols] for r in rows]
    SH.overwrite(spec["tab"], cols, grid)
    return len(grid)


# Serialize all mirror writes through one worker so concurrent mutations can't
# race on the shared Sheets client / token.
_mirror_q = queue.Queue()


def _mirror_worker():
    while True:
        entity = _mirror_q.get()
        try:
            mirror_entity(entity)
        except Exception:  # noqa: BLE001
            print(f"[mirror] {entity} sync failed:", flush=True)
            traceback.print_exc()
        finally:
            _mirror_q.task_done()


threading.Thread(target=_mirror_worker, daemon=True).start()


def mirror_async(entity):
    if SH is not None:
        _mirror_q.put(entity)

# Static dashboard, with the CRM token injected so the (basic-auth-gated)
# operator page can call the token-protected mutation endpoints.
with open(os.path.join(os.path.dirname(__file__), "static", "index.html")) as _f:
    INDEX_HTML = _f.read().replace("__CRM_TOKEN__", CRM_TOKEN)


def authed():
    return bool(CRM_TOKEN) and request.headers.get("X-CRM-Token") == CRM_TOKEN


def log_activity(cur, client_id, action, detail, result="ok"):
    cur.execute(
        "INSERT INTO activity_log (ts, workflow, client_id, action, detail, result) "
        "VALUES (now(), %s, %s, %s, %s, %s)",
        ("back-office", client_id, action, detail, result))

LIST_ORDER = {
    "leads": "received_at DESC NULLS LAST",
    "clients": "client_id",
    "projects": "project_id",
    "bookings": "start_time DESC NULLS LAST",
    "invoices": "issued_date DESC NULLS LAST",
    "activity_log": "ts DESC NULLS LAST",
}


def jsonable(v):
    if isinstance(v, (datetime, date)):
        return v.isoformat()
    if isinstance(v, Decimal):
        return float(v)
    return v


def rows_json(rows):
    return [{k: jsonable(v) for k, v in r.items()} for r in rows]


@app.get("/healthz")
def healthz():
    try:
        with db.connect() as conn, conn.cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()
        return Response("ok\n", mimetype="text/plain")
    except Exception as e:  # noqa: BLE001
        return Response(f"db error: {e}\n", status=500, mimetype="text/plain")


@app.get("/api/<entity>")
def list_entity(entity):
    if entity not in db.TABLES:
        return jsonify({"error": "unknown entity"}), 404
    order = LIST_ORDER.get(entity, db.TABLES[entity]["pk"])
    with db.connect() as conn, conn.cursor() as cur:
        cur.execute(f"SELECT * FROM {entity} ORDER BY {order}")
        rows = cur.fetchall()
    return jsonify({"entity": entity, "count": len(rows), "rows": rows_json(rows)})


def compute_renewal(start, cycle):
    """Mirror the onboarding workflow: monthly -> +1 month, yearly -> +1 year."""
    cycle = (cycle or "monthly").lower()
    if cycle == "monthly":
        m = start.month % 12 + 1
        y = start.year + (1 if start.month == 12 else 0)
        d = min(start.day, [31, 29 if y % 4 == 0 and (y % 100 or not y % 400) else 28,
                            31, 30, 31, 30, 31, 31, 30, 31, 30, 31][m - 1])
        return date(y, m, d)
    if cycle == "yearly":
        return date(start.year + 1, start.month, start.day)
    return None


def next_client_id(cur):
    cur.execute("SELECT client_id FROM clients")
    mx = 0
    for r in cur.fetchall():
        m = re.match(r"C-(\d+)$", r["client_id"] or "")
        if m:
            mx = max(mx, int(m.group(1)))
    return "C-%04d" % (mx + 1)


@app.post("/api/<entity>")
def add_entity(entity):
    if entity not in db.TABLES:
        return jsonify({"error": "unknown entity"}), 404
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    spec = db.TABLES[entity]
    pk = spec["cols"][0] if entity == "activity_log" else spec["pk"]
    body = request.get_json(force=True, silent=True) or {}
    row = db.coerce_row(entity, body)
    with db.connect() as conn, conn.cursor() as cur:
        if entity == "leads":
            row["lead_id"] = row.get("lead_id") or "L-" + str(int(time.time() * 1000))
            row["received_at"] = row.get("received_at") or datetime.now(timezone.utc)
            row["status"] = row.get("status") or "new"
            if row.get("notified") is None:
                row["notified"] = False
        elif entity == "clients":
            row["client_id"] = row.get("client_id") or next_client_id(cur)
            row["created_at"] = row.get("created_at") or datetime.now(timezone.utc)
            row["status"] = row.get("status") or "lead"
            if not row.get("renewal_date") and row.get("start_date"):
                row["renewal_date"] = compute_renewal(
                    row["start_date"], row.get("billing_cycle"))
        elif not row.get(pk):
            return jsonify({"error": f"{pk} required"}), 400
        cols = spec["cols"]
        ph = ", ".join(["%s"] * len(cols))
        cur.execute(f"INSERT INTO {entity} ({', '.join(cols)}) VALUES ({ph})",
                    [row.get(c) for c in cols])
        log_activity(cur, row.get("client_id"), f"add {entity}", f"{pk}={row.get(pk)}")
        conn.commit()
    mirror_async(entity)
    mirror_async("activity_log")
    return jsonify({"added": row.get(pk)}), 201


@app.patch("/api/<entity>/<path:ident>")
def edit_entity(entity, ident):
    if entity not in db.TABLES:
        return jsonify({"error": "unknown entity"}), 404
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    spec = db.TABLES[entity]
    pk = spec["pk"]
    body = request.get_json(force=True, silent=True) or {}
    setcols = [c for c in body if c in spec["cols"] and c != pk]
    if not setcols:
        return jsonify({"error": "no editable fields"}), 400
    typed = db.coerce_row(entity, body)
    setsql = ", ".join(f"{c} = %s" for c in setcols)
    vals = [typed[c] for c in setcols] + [ident]
    with db.connect() as conn, conn.cursor() as cur:
        cur.execute(f"UPDATE {entity} SET {setsql} WHERE {pk} = %s RETURNING {pk}", vals)
        if not cur.fetchone():
            return jsonify({"error": "not found"}), 404
        log_activity(cur, ident if entity == "clients" else None,
                     f"edit {entity}", f"{pk}={ident}: {', '.join(setcols)}")
        conn.commit()
    mirror_async(entity)
    mirror_async("activity_log")
    return jsonify({"updated": ident, "fields": setcols})


@app.delete("/api/<entity>/<path:ident>")
def delete_entity(entity, ident):
    if entity not in db.TABLES:
        return jsonify({"error": "unknown entity"}), 404
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    pk = db.TABLES[entity]["pk"]
    with db.connect() as conn, conn.cursor() as cur:
        cur.execute(f"DELETE FROM {entity} WHERE {pk} = %s RETURNING {pk}", (ident,))
        if not cur.fetchone():
            return jsonify({"error": "not found"}), 404
        log_activity(cur, ident if entity == "clients" else None,
                     f"delete {entity}", f"{pk}={ident}")
        conn.commit()
    mirror_async(entity)
    mirror_async("activity_log")
    return jsonify({"deleted": ident})


@app.post("/api/sync")
def sync_all():
    """Full DB -> Sheets resync of every tab (manual / alignment)."""
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    if SH is None:
        return jsonify({"error": "sheets unavailable"}), 503
    out = {}
    for entity in db.TABLES:
        try:
            out[entity] = mirror_entity(entity)
        except Exception as e:  # noqa: BLE001
            out[entity] = f"error: {e}"
    return jsonify({"synced": out})


def _ics_dt(v):
    if isinstance(v, datetime):
        u = v if v.tzinfo else v.replace(tzinfo=timezone.utc)
        return u.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return None


def _ics_esc(t):
    return (str(t or "").replace("\\", "\\\\").replace(";", "\\;")
            .replace(",", "\\,").replace("\n", "\\n"))


@app.get("/api/bookings.ics")
def bookings_ics():
    """Read-only iCal feed for Apple/Google Calendar subscription. Gated by the
    ICS_TOKEN query param (no header auth, so calendar apps can fetch it)."""
    if not ICS_TOKEN or request.args.get("token") != ICS_TOKEN:
        return Response("forbidden\n", status=403, mimetype="text/plain")
    cid = request.args.get("client_id")
    with db.connect() as conn, conn.cursor() as cur:
        if cid:
            cur.execute("SELECT * FROM bookings WHERE client_id = %s ORDER BY start_time", (cid,))
        else:
            cur.execute("SELECT * FROM bookings ORDER BY start_time")
        rows = cur.fetchall()
    now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    out = ["BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//smb-crm//bookings//DE",
           "CALSCALE:GREGORIAN", "METHOD:PUBLISH",
           "X-WR-CALNAME:" + _ics_esc("Buchungen " + cid if cid else "Buchungen")]
    for r in rows:
        st = _ics_dt(r.get("start_time"))
        if not st:
            continue
        summary = r.get("service") or "Termin"
        if r.get("customer_name"):
            summary += " – " + r["customer_name"]
        out += ["BEGIN:VEVENT",
                "UID:%s@smb-crm" % (r.get("booking_id") or now),
                "DTSTAMP:%s" % (_ics_dt(r.get("created_at")) or now),
                "DTSTART:%s" % st]
        en = _ics_dt(r.get("end_time"))
        if en:
            out.append("DTEND:%s" % en)
        out.append("SUMMARY:" + _ics_esc(summary))
        if r.get("customer_contact"):
            out.append("DESCRIPTION:" + _ics_esc("Kontakt: " + r["customer_contact"]))
        out += ["STATUS:CONFIRMED", "END:VEVENT"]
    out.append("END:VCALENDAR")
    return Response("\r\n".join(out) + "\r\n", mimetype="text/calendar")


@app.get("/")
def index():
    return Response(INDEX_HTML, mimetype="text/html")


if __name__ == "__main__":
    serve(app, host="0.0.0.0", port=8080)