"""Feedback submission service. Never log user text, uploads, or SMTP secrets."""

import asyncio
import hashlib
import hmac
import json
import os
import re
import smtplib
import sqlite3
import ssl
import tempfile
import threading
import time
from collections import deque
from contextlib import asynccontextmanager, contextmanager, suppress
from dataclasses import dataclass, field
from email.message import EmailMessage
from email.policy import SMTP
from email.utils import formatdate
from pathlib import Path

from fastapi import FastAPI, Request
from starlette.datastructures import UploadFile
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse
from starlette.concurrency import run_in_threadpool

MIB = 1024 * 1024
FILE_LIMIT = 8 * MIB
TOTAL_FILE_LIMIT = 20 * MIB
BODY_LIMIT = TOTAL_FILE_LIMIT + 64 * 1024
ID_PATTERN = re.compile(r"[A-Za-z0-9_-]{16,128}\Z")
ADDRESS_PATTERN = re.compile(r"[A-Za-z0-9._%+-]+@(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,63}\Z")


@dataclass
class Settings:
    database_path: str = "data/feedback.sqlite3"
    smtp_host: str = "smtp.feishu.cn"
    smtp_port: int = 465
    smtp_tls: str = "ssl"
    smtp_username: str = ""
    smtp_password: str = field(default="", repr=False)
    mail_from: str = ""
    mail_to: str = ""
    idempotency_secret: str = field(default="", repr=False)
    smtp_timeout_seconds: int = 15
    retention_days: int = 7
    inflight_stale_seconds: int = 300
    ip_requests_per_hour: int = 10
    global_requests_per_hour: int = 100
    max_concurrent_requests: int = 4
    upload_timeout_seconds: int = 60

    @classmethod
    def from_env(cls):
        names = {
            "database_path": "DATABASE_PATH", "smtp_host": "SMTP_HOST", "smtp_port": "SMTP_PORT",
            "smtp_tls": "SMTP_TLS", "smtp_username": "SMTP_USERNAME", "smtp_password": "SMTP_PASSWORD",
            "mail_from": "MAIL_FROM", "mail_to": "MAIL_TO", "idempotency_secret": "IDEMPOTENCY_SECRET",
            "smtp_timeout_seconds": "SMTP_TIMEOUT_SECONDS", "retention_days": "RETENTION_DAYS",
            "ip_requests_per_hour": "IP_REQUESTS_PER_HOUR", "global_requests_per_hour": "GLOBAL_REQUESTS_PER_HOUR",
            "max_concurrent_requests": "MAX_CONCURRENT_REQUESTS", "upload_timeout_seconds": "UPLOAD_TIMEOUT_SECONDS",
        }
        defaults, values = cls(), {}
        for name, env_name in names.items():
            if env_name in os.environ:
                value = os.environ[env_name]
                if isinstance(getattr(defaults, name), int):
                    try:
                        value = int(value)
                    except ValueError:
                        raise ValueError(f"{env_name} must be an integer") from None
                values[name] = value
        return cls(**values)

    def validate(self):
        for name in ("mail_from", "mail_to"):
            if not ADDRESS_PATTERN.fullmatch(getattr(self, name)):
                raise ValueError(f"{name} must be one plain email address")
        if not self.smtp_username.strip() or "\r" in self.smtp_username or "\n" in self.smtp_username:
            raise ValueError("SMTP_USERNAME is required and must not contain newlines")
        if not self.smtp_password or self.smtp_password.startswith("REPLACE_"):
            raise ValueError("SMTP_PASSWORD is required")
        if not re.fullmatch(r"[A-Za-z0-9.-]{1,253}", self.smtp_host):
            raise ValueError("SMTP_HOST must be a hostname")
        if (self.smtp_port, self.smtp_tls) not in {(465, "ssl"), (587, "starttls")}:
            raise ValueError("Use SMTP_PORT=465/SMTP_TLS=ssl or SMTP_PORT=587/SMTP_TLS=starttls")
        if len(self.idempotency_secret.encode()) < 32 or self.idempotency_secret.startswith("REPLACE_"):
            raise ValueError("IDEMPOTENCY_SECRET must have at least 32 bytes")
        for name in ("retention_days", "smtp_timeout_seconds", "inflight_stale_seconds",
                     "ip_requests_per_hour", "global_requests_per_hour", "max_concurrent_requests",
                     "upload_timeout_seconds"):
            if getattr(self, name) <= 0:
                raise ValueError(f"{name} must be positive")
        if self.smtp_timeout_seconds > 30 or self.max_concurrent_requests > 16:
            raise ValueError("SMTP timeout must be <=30 seconds and concurrency <=16")


class ApiError(Exception):
    def __init__(self, http_status, error, detail, *, submission_id=None, status=None):
        self.http_status, self.error, self.detail = http_status, error, detail
        self.submission_id, self.status = submission_id, status


class SendFailure(Exception):
    """SMTP definitely did not accept this message. Retrying is safe."""


class UncertainFailure(Exception):
    """DATA may have been accepted, but its acknowledgement was lost."""


def error_response(code, error, detail, *, submission_id=None, status=None, headers=None):
    data = {"success": False, "error": error, "detail": detail}
    if submission_id is not None:
        data["submission_id"] = submission_id
    if status is not None:
        data["status"] = status
    return JSONResponse(data, status_code=code, headers=headers)


class SmtpSender:
    def __init__(self, settings):
        self.settings = settings

    def __call__(self, message):
        config, client, in_data, accepted = self.settings, None, False, False
        raw_message = message.as_bytes(policy=SMTP)
        try:
            context = ssl.create_default_context()
            if config.smtp_tls == "ssl":
                client = smtplib.SMTP_SSL(config.smtp_host, config.smtp_port,
                                          timeout=config.smtp_timeout_seconds, context=context)
            else:
                client = smtplib.SMTP(config.smtp_host, config.smtp_port, timeout=config.smtp_timeout_seconds)
                client.ehlo()
                client.starttls(context=context)
                client.ehlo()
            client.login(config.smtp_username, config.smtp_password)
            code, _ = client.mail(config.mail_from)
            if code != 250:
                raise SendFailure()
            code, _ = client.rcpt(config.mail_to)
            if code not in (250, 251):
                raise SendFailure()
            # A lost response from DATA is ambiguous: never automatically resend.
            in_data = True
            code, _ = client.data(raw_message)
            if code != 250:
                raise SendFailure()
            accepted = True
        except SendFailure:
            raise
        except smtplib.SMTPDataError:
            raise SendFailure() from None
        except (OSError, smtplib.SMTPException):
            if in_data:
                raise UncertainFailure() from None
            raise SendFailure() from None
        finally:
            if client is not None:
                try:
                    if accepted:
                        client.quit()
                    else:
                        client.close()
                except (OSError, smtplib.SMTPException):
                    # QUIT failures do not undo an acknowledged DATA acceptance.
                    client.close()


class SubmissionStore:
    def __init__(self, settings):
        self.settings = settings
        Path(settings.database_path).parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as db:
            db.execute("PRAGMA journal_mode=WAL")
            db.execute("CREATE TABLE IF NOT EXISTS submissions ("
                       "submission_id TEXT PRIMARY KEY, digest TEXT NOT NULL, status TEXT NOT NULL, "
                       "created REAL NOT NULL, updated REAL NOT NULL)")
            db.execute("CREATE INDEX IF NOT EXISTS submissions_updated ON submissions(updated)")
        self.prune()

    @contextmanager
    def connect(self):
        db = sqlite3.connect(self.settings.database_path, timeout=5)
        try:
            db.execute("PRAGMA synchronous=FULL")
            db.execute("PRAGMA secure_delete=ON")
            with db:
                yield db
        finally:
            db.close()

    def claim(self, submission_id, digest):
        now = time.time()
        with self.connect() as db:
            db.execute("BEGIN IMMEDIATE")
            db.execute("DELETE FROM submissions WHERE updated < ?",
                       (now - self.settings.retention_days * 86400,))
            row = db.execute("SELECT digest, status, updated FROM submissions WHERE submission_id=?",
                             (submission_id,)).fetchone()
            if row:
                old_digest, status, updated = row
                if not hmac.compare_digest(old_digest, digest):
                    return "conflict"
                if status == "sending":
                    if now - updated <= self.settings.inflight_stale_seconds:
                        return "in_progress"
                    db.execute("UPDATE submissions SET status='uncertain', updated=? WHERE submission_id=?",
                               (now, submission_id))
                    return "uncertain"
                if status in ("sent", "uncertain"):
                    return status
                db.execute("UPDATE submissions SET status='sending', updated=? WHERE submission_id=?",
                           (now, submission_id))
            else:
                db.execute("INSERT INTO submissions VALUES (?, ?, 'sending', ?, ?)",
                           (submission_id, digest, now, now))
        return "claimed"

    def finish(self, submission_id, status):
        with self.connect() as db:
            db.execute("UPDATE submissions SET status=?, updated=? WHERE submission_id=?",
                       (status, time.time(), submission_id))

    def prune(self):
        with self.connect() as db:
            db.execute("DELETE FROM submissions WHERE updated < ?",
                       (time.time() - self.settings.retention_days * 86400,))


class AdmissionControl:
    """Bounded per-process admission state: run exactly one worker."""
    def __init__(self, settings):
        self.settings, self.lock, self.current = settings, threading.Lock(), 0
        self.global_times, self.ip_times = deque(), {}

    def enter(self, ip):
        now = time.monotonic()
        ip_key = hmac.new(self.settings.idempotency_secret.encode(), ip.encode(), hashlib.sha256).digest()
        with self.lock:
            while self.global_times and self.global_times[0] <= now - 3600:
                self.global_times.popleft()
            self.ip_times = {k: times for k, times in self.ip_times.items() if times[-1] > now - 3600}
            times = self.ip_times.get(ip_key, deque())
            while times and times[0] <= now - 3600:
                times.popleft()
            if (len(times) >= self.settings.ip_requests_per_hour or
                    len(self.global_times) >= self.settings.global_requests_per_hour):
                return error_response(429, "rate_limited", "Submission limit reached; try again later.",
                                      headers={"Retry-After": "3600"})
            if self.current >= self.settings.max_concurrent_requests:
                return error_response(503, "server_busy", "Server is busy; retry with the same submission ID.",
                                      headers={"Retry-After": "15"})
            self.ip_times[ip_key] = times
            times.append(now)
            self.global_times.append(now)
            self.current += 1
        return None

    def leave(self):
        with self.lock:
            self.current -= 1


class BoundedFeedbackBody:
    """Count raw bytes before parsing, including chunked bodies and multipart overhead."""
    def __init__(self, app, settings, admission):
        self.app, self.settings, self.admission = app, settings, admission

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or scope.get("path") != "/v1/feedback" or scope["method"] != "POST":
            return await self.app(scope, receive, send)
        denial = self.admission.enter((scope.get("client") or ("unknown",))[0])
        if denial:
            return await denial(scope, receive, send)
        try:
            headers = dict(scope.get("headers", []))
            if headers.get(b"content-type", b"").split(b";", 1)[0].strip().lower() != b"multipart/form-data":
                return await error_response(415, "unsupported_content_type", "Use multipart/form-data.")(scope, receive, send)
            if b"content-length" in headers:
                try:
                    length = int(headers[b"content-length"])
                except ValueError:
                    return await error_response(400, "invalid_body", "Invalid Content-Length.")(scope, receive, send)
                if length < 0:
                    return await error_response(400, "invalid_body", "Invalid Content-Length.")(scope, receive, send)
                if length > BODY_LIMIT:
                    return await error_response(413, "upload_too_large", "Request exceeds upload limits.")(scope, receive, send)
            with tempfile.SpooledTemporaryFile(max_size=MIB, mode="w+b") as spool:
                total, deadline = 0, time.monotonic() + self.settings.upload_timeout_seconds
                while True:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        return await error_response(408, "upload_timeout", "Upload timed out.")(scope, receive, send)
                    try:
                        event = await asyncio.wait_for(receive(), remaining)
                    except TimeoutError:
                        return await error_response(408, "upload_timeout", "Upload timed out.")(scope, receive, send)
                    if event["type"] == "http.disconnect":
                        return
                    body = event.get("body", b"")
                    total += len(body)
                    if total > BODY_LIMIT:
                        return await error_response(413, "upload_too_large", "Request exceeds upload limits.")(scope, receive, send)
                    spool.write(body)
                    if not event.get("more_body", False):
                        break
                spool.seek(0)
                replayed = False

                async def replay_receive():
                    nonlocal replayed
                    if replayed:
                        return {"type": "http.disconnect"}
                    block = spool.read(64 * 1024)
                    replayed = spool.tell() >= total
                    return {"type": "http.request", "body": block, "more_body": not replayed}

                return await self.app(scope, replay_receive, send)
        finally:
            self.admission.leave()


def file_signature(data, mime):
    if mime == "image/jpeg":
        return data.startswith(b"\xff\xd8\xff"), ".jpg"
    if mime == "image/png":
        return data.startswith(b"\x89PNG\r\n\x1a\n"), ".png"
    if mime == "image/gif":
        return data.startswith((b"GIF87a", b"GIF89a")), ".gif"
    if mime == "image/webp":
        return len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP", ".webp"
    if mime == "video/webm":
        return data.startswith(b"\x1a\x45\xdf\xa3") and b"webm" in data[:4096], ".webm"
    if mime in {"image/heic", "image/heif", "video/mp4", "video/quicktime", "video/x-m4v"}:
        if len(data) < 16 or data[4:8] != b"ftyp":
            return False, ""
        box_size = int.from_bytes(data[:4], "big")
        if box_size < 16 or box_size > len(data) or box_size > 4096 or box_size % 4:
            return False, ""
        brands = {data[8:12]} | {data[i:i + 4] for i in range(16, box_size, 4)}
        if mime in {"image/heic", "image/heif"}:
            return bool(brands & {b"heic", b"heix", b"hevc", b"hevx", b"heim", b"heis", b"hevm", b"hevs"}), ".heic"
        if mime == "video/quicktime":
            return b"qt  " in brands, ".mov"
        return bool(brands & {b"isom", b"iso2", b"avc1", b"mp41", b"mp42", b"M4V ", b"M4VH", b"M4VP", b"dash", b"MSNV"}), ".mp4"
    return False, ""


def safe_filename(filename, extension):
    filename = (filename or "attachment").replace("\\", "/").rsplit("/", 1)[-1]
    filename = re.sub(r"[^A-Za-z0-9._-]", "_", filename).strip("._")[:100] or "attachment"
    alternatives = {".jpg": {".jpg", ".jpeg"}, ".heic": {".heic", ".heif"}, ".mp4": {".mp4", ".m4v"}}
    if Path(filename).suffix.lower() not in alternatives.get(extension, {extension}):
        filename = Path(filename).stem + extension
    return filename


async def parse_submission(request):
    try:
        async with request.form(max_files=3, max_fields=5, max_part_size=20000) as form:
            fields, uploads = {}, []
            for name, value in form.multi_items():
                if isinstance(value, UploadFile):
                    if name != "attachments":
                        raise ApiError(422, "validation_error", "Unexpected file field.")
                    uploads.append(value)
                else:
                    if name not in {"submission_id", "type", "message", "contact", "school"} or name in fields:
                        raise ApiError(422, "validation_error", "Unexpected or duplicated form field.")
                    fields[name] = value
            submission_id = fields.get("submission_id", "")
            if not ID_PATTERN.fullmatch(submission_id):
                raise ApiError(422, "validation_error", "submission_id must be a random 16–128 character identifier.")
            kind = fields.get("type")
            if kind not in {"problem", "development"}:
                raise ApiError(422, "validation_error", "type must be problem or development.")
            for name, maximum in (("message", 5000), ("contact", 200), ("school", 200)):
                if len(fields.get(name, "")) > maximum:
                    raise ApiError(422, "validation_error", f"{name} exceeds {maximum} characters.")
            if any("\x00" in text for text in fields.values()):
                raise ApiError(422, "validation_error", "Null characters are not allowed.")
            if kind == "problem":
                if not fields.get("message", "").strip() or "school" in fields:
                    raise ApiError(422, "validation_error", "Problem feedback requires message and must omit school.")
            elif ("message" in fields or uploads or not fields.get("school", "").strip()
                  or not fields.get("contact", "").strip()):
                raise ApiError(422, "validation_error", "Development requests require school and contact; omit message and attachments.")
            attachments, total = [], 0
            for upload in uploads:
                data = await upload.read(FILE_LIMIT + 1)
                total += len(data)
                if len(data) > FILE_LIMIT or total > TOTAL_FILE_LIMIT:
                    raise ApiError(413, "upload_too_large", "Maximum three files, 8 MiB each and 20 MiB total.")
                mime = (upload.content_type or "").lower()
                valid, extension = file_signature(data, mime)
                if not valid:
                    raise ApiError(415, "unsupported_media", "Attachment MIME type and file signature must match a supported image or video.")
                attachments.append((safe_filename(upload.filename, extension), mime, data))
            return fields, attachments
    except HTTPException as exc:
        detail = str(exc.detail).lower()
        if "too many" in detail or "maximum" in detail or "exceeded maximum size" in detail:
            raise ApiError(413, "upload_too_large", "Multipart field or upload limits exceeded.") from None
        raise ApiError(400, "invalid_body", "Malformed multipart body.") from None


def make_message(settings, fields, attachments):
    message = EmailMessage(policy=SMTP)
    message["From"], message["To"] = settings.mail_from, settings.mail_to
    message["Subject"] = f"[App feedback] {fields['type']} / {fields['submission_id']}"
    message["Message-ID"] = f"<{fields['submission_id']}@{settings.mail_from.split('@')[1]}>"
    message["Date"] = formatdate(localtime=False, usegmt=True)
    lines = [f"Submission ID: {fields['submission_id']}", f"Type: {fields['type']}"]
    if fields["type"] == "development":
        lines += ["School:", fields["school"], "Contact:", fields["contact"]]
    else:
        lines += ["Contact:", fields.get("contact", ""), "Message:", fields["message"]]
    message.set_content("\n".join(lines), charset="utf-8")
    for name, mime, data in attachments:
        main, sub = mime.split("/", 1)
        message.add_attachment(data, maintype=main, subtype=sub, filename=name)
    return message


def create_app(settings=None, sender=None):
    settings = settings or Settings.from_env()
    settings.validate()
    store = SubmissionStore(settings)

    @asynccontextmanager
    async def lifespan(app):
        async def cleanup():
            while True:
                await asyncio.sleep(3600)
                # Persistent storage failure makes readiness fail instead of exposing data in logs.
                try:
                    await run_in_threadpool(store.prune)
                    app.state.retention_healthy = True
                except sqlite3.Error:
                    app.state.retention_healthy = False

        app.state.retention_healthy = True
        task = asyncio.create_task(cleanup())
        try:
            yield
        finally:
            task.cancel()
            with suppress(asyncio.CancelledError):
                await task

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
    app.state.sender = sender or SmtpSender(settings)
    app.add_middleware(BoundedFeedbackBody, settings=settings, admission=AdmissionControl(settings))

    @app.exception_handler(ApiError)
    async def api_error_handler(request, exc):
        return error_response(exc.http_status, exc.error, exc.detail,
                              submission_id=exc.submission_id, status=exc.status)

    @app.get("/healthz")
    async def health():
        # Liveness only: never sends email or claims SMTP readiness.
        if not getattr(app.state, "retention_healthy", True):
            return error_response(503, "storage_unavailable", "Retention cleanup requires operator attention.")
        return {"status": "ok"}

    @app.post("/v1/feedback")
    async def feedback(request: Request):
        fields, attachments = await parse_submission(request)
        submission_id = fields["submission_id"]
        canonical = json.dumps({"fields": fields, "attachments": [
            {"name": name, "mime": mime, "sha256": hashlib.sha256(data).hexdigest()}
            for name, mime, data in attachments]}, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
        digest = hmac.new(settings.idempotency_secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
        message = make_message(settings, fields, attachments)

        def deliver():
            try:
                outcome = store.claim(submission_id, digest)
                if outcome == "conflict":
                    raise ApiError(409, "id_conflict", "This ID was already used with different content.", submission_id=submission_id)
                if outcome == "in_progress":
                    raise ApiError(409, "in_progress", "A submission with this ID is still being processed.",
                                   submission_id=submission_id, status="in_progress")
                if outcome == "uncertain":
                    raise ApiError(503, "delivery_uncertain", "Mail acceptance is uncertain; do not submit a new ID. Contact the operator.",
                                   submission_id=submission_id, status="uncertain")
                if outcome == "sent":
                    return
                try:
                    app.state.sender(message)
                except SendFailure:
                    store.finish(submission_id, "retryable")
                    raise ApiError(503, "smtp_unavailable", "Mail was not accepted; retry with the same ID.",
                                   submission_id=submission_id, status="retryable") from None
                except Exception:
                    store.finish(submission_id, "uncertain")
                    raise ApiError(503, "delivery_uncertain", "Mail acceptance is uncertain; do not submit a new ID. Contact the operator.",
                                   submission_id=submission_id, status="uncertain") from None
                store.finish(submission_id, "sent")
            except sqlite3.Error:
                raise ApiError(503, "storage_unavailable", "Submission state could not be confirmed; retry only with the same ID.",
                               submission_id=submission_id, status="uncertain") from None

        await run_in_threadpool(deliver)
        return {"success": True, "submission_id": submission_id, "status": "sent"}

    return app