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 booking_db as bdb
import db
import owner_mail
from sheets import Sheets
from booking_api import bp as booking_bp
from public_booking import bp as public_booking_bp
from manage_booking import bp as manage_booking_bp
from owner_auth import bp as owner_auth_bp
from owner_booking import bp as owner_booking_bp

app = Flask(__name__, static_folder="static", static_url_path="")
app.register_blueprint(booking_bp)
app.register_blueprint(public_booking_bp)
app.register_blueprint(manage_booking_bp)
app.register_blueprint(owner_auth_bp)
app.register_blueprint(owner_booking_bp)

# Dedicated secret for the owner-login session cookie -- deliberately not
# shared with CRM_API_TOKEN or BOOKING_TOKEN_SECRET (#19), same reasoning as
# booking_api.py's own token secret: rotating one must never silently affect
# the others.
app.secret_key = os.environ.get("SESSION_SECRET_KEY", "")

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):
    spec = db.TABLES[entity]
    if SH is None or spec.get("mirror") is False:
        return 0
    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 and db.TABLES[entity].get("mirror") is not False:
        _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",
    "credentials": "client_id",
}


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
    # Credentials carry secrets, not just business metadata — require the
    # mutation-level token even to read, on top of the Caddy basic-auth every
    # other entity's (read-only) list relies on alone.
    if entity == "credentials" and not authed():
        return jsonify({"error": "forbidden"}), 403
    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 entity == "credentials":
            row["cred_id"] = row.get("cred_id") or "CR-" + str(int(time.time() * 1000))
            row["created_at"] = row.get("created_at") or datetime.now(timezone.utc)
        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.get("/api/owner_users")
def list_owner_users():
    """Owner-accounts list for the CRM dashboard's new tab (#19). Requires
    the CRM token to read, same as credentials -- these rows identify who can
    log into a client's owner portal."""
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    rows = bdb.list_users()
    return jsonify({"entity": "owner_users", "count": len(rows), "rows": rows_json(rows)})


@app.post("/api/owner_users/<user_id>/reset-password")
def trigger_owner_password_reset(user_id):
    """Operator-triggered reset (#19): sends the same reset email an owner
    would send themselves via /owner/forgot-password."""
    if not authed():
        return jsonify({"error": "forbidden"}), 403
    user = bdb.get_user(user_id)
    if user is None:
        return jsonify({"error": "not found"}), 404
    client = bdb.get_client(user["client_id"])
    tok = bdb.create_password_reset_token(user["user_id"])
    owner_mail.send_password_reset(client, user, tok["token"])
    with db.connect() as conn, conn.cursor() as cur:
        log_activity(cur, user["client_id"], "owner password reset",
                     f"user_id={user_id} (operator-triggered)")
        conn.commit()
    mirror_async("activity_log")
    return jsonify({"sent": user_id})


@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)