#!/usr/bin/env python3
"""Educational model, not a cloud provider. Python standard library only.
SQLite is both the durable operation journal and a fake external provider.
The fake provider commits independently to reproduce a lost response.
No Kubernetes, SSH or OpenStack calls. HTTP binds loopback only.
"""
import argparse
import contextlib
import hmac
import json
import os
from pathlib import Path
import secrets
import sqlite3
import tempfile
import uuid
from concurrent.futures import ThreadPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Rejected(Exception):
    def __init__(self, status, message):
        self.status, self.message = status, message

def connect(path):
    db = sqlite3.connect(path, timeout=10, isolation_level=None)
    db.row_factory = sqlite3.Row
    db.executescript("""
    CREATE TABLE IF NOT EXISTS clusters(
      id TEXT PRIMARY KEY, owner TEXT NOT NULL, plan TEXT NOT NULL, state TEXT NOT NULL);
    CREATE TABLE IF NOT EXISTS operations(
      id TEXT PRIMARY KEY, owner TEXT NOT NULL, cluster_id TEXT NOT NULL,
      kind TEXT NOT NULL, state TEXT NOT NULL);
    CREATE TABLE IF NOT EXISTS requests(
      owner TEXT NOT NULL, key TEXT NOT NULL, payload TEXT NOT NULL,
      op_id TEXT NOT NULL, PRIMARY KEY(owner,key));
    CREATE TABLE IF NOT EXISTS cloud(cluster_id TEXT PRIMARY KEY);
    """)
    return db

@contextlib.contextmanager
def transaction(path):
    db = connect(path)
    try:
        db.execute("BEGIN IMMEDIATE")
        yield db
        db.commit()
    except BaseException:
        db.rollback()
        raise
    finally:
        db.close()

def request_create(path, owner, key, plan):
    if not key or len(key)>128 or plan not in ("small", "large"):
        raise Rejected(400, "key and supported plan required")
    payload=json.dumps({"plan":plan}, sort_keys=True)
    with transaction(path) as db:
        old=db.execute("SELECT * FROM requests WHERE owner=? AND key=?", (owner,key)).fetchone()
        if old:
            if old["payload"]!=payload:
                raise Rejected(409, "same key with different payload")
            return dict(db.execute("SELECT * FROM operations WHERE id=?", (old["op_id"],)).fetchone())
        count=db.execute("SELECT count(*) FROM clusters WHERE owner=? AND state!='Deleted'",(owner,)).fetchone()[0]
        if count>=1:
            raise Rejected(409, "tenant quota reserved; limit is one active cluster")
        cid,oid=str(uuid.uuid4()),str(uuid.uuid4())
        db.execute("INSERT INTO clusters VALUES(?,?,?,'Pending')",(cid,owner,plan))
        db.execute("INSERT INTO operations VALUES(?,?,?,'create','Pending')",(oid,owner,cid))
        db.execute("INSERT INTO requests VALUES(?,?,?,?)",(owner,key,payload,oid))
        return dict(db.execute("SELECT * FROM operations WHERE id=?",(oid,)).fetchone())

def owned(path, table, owner, identity):
    assert table in ("clusters","operations")
    with contextlib.closing(connect(path)) as db:
        row=db.execute("SELECT * FROM "+table+" WHERE id=? AND owner=?",(identity,owner)).fetchone()
        if not row:
            raise Rejected(404,"object not found")
        return dict(row)

def request_delete(path, owner, cid):
    with transaction(path) as db:
        row=db.execute("SELECT * FROM clusters WHERE id=? AND owner=?",(cid,owner)).fetchone()
        if not row:
            raise Rejected(404,"object not found")
        old=db.execute("SELECT * FROM operations WHERE cluster_id=? AND kind='delete'",(cid,)).fetchone()
        if old:
            return dict(old)
        if row["state"]!="Ready":
            raise Rejected(409,"wait for create to complete")
        oid=str(uuid.uuid4())
        db.execute("INSERT INTO operations VALUES(?,?,?,'delete','Pending')",(oid,owner,cid))
        db.execute("UPDATE clusters SET state='Deleting' WHERE id=?",(cid,))
        return dict(db.execute("SELECT * FROM operations WHERE id=?",(oid,)).fetchone())

def reconcile(path, crash=False):
    with contextlib.closing(connect(path)) as db:
        pending=[dict(r) for r in db.execute("SELECT * FROM operations WHERE state='Pending' ORDER BY rowid")]
    for op in pending:
        # A stable cluster_id identifies the same external resource after restart.
        with transaction(path) as db:
            if op["kind"]=="create":
                db.execute("INSERT OR IGNORE INTO cloud VALUES(?)",(op["cluster_id"],))
            else:
                db.execute("DELETE FROM cloud WHERE cluster_id=?",(op["cluster_id"],))
        if crash:
            raise Rejected(75,"simulated crash after provider commit; operation is still Pending")
        with transaction(path) as db:
            db.execute("UPDATE operations SET state='Succeeded' WHERE id=?",(op["id"],))
            db.execute("UPDATE clusters SET state=? WHERE id=?",
                       ("Ready" if op["kind"]=="create" else "Deleted",op["cluster_id"]))

def inventory(path,owner):
    with contextlib.closing(connect(path)) as db:
        return {"clusters":[dict(r) for r in db.execute("SELECT * FROM clusters WHERE owner=?",(owner,))],
                "operations":[dict(r) for r in db.execute("SELECT * FROM operations WHERE owner=?",(owner,))],
                "external_resources":[r[0] for r in db.execute(
                  "SELECT cloud.cluster_id FROM cloud JOIN clusters ON clusters.id=cloud.cluster_id WHERE owner=?",(owner,))]}

def token_file(path):
    return Path(str(path)+".tokens.json")

def init_tokens(path):
    values={name:secrets.token_urlsafe(32) for name in ("alice","bob")}
    with token_file(path).open("x",encoding="utf-8") as out:
        json.dump(values,out)
    return {"credentials_file":str(token_file(path)), "note":"mode 600; never publish its contents"}

def make_server(path, address=("127.0.0.1",18080)):
    tokens=json.loads(token_file(path).read_text())
    class Handler(BaseHTTPRequestHandler):
        def log_message(self,*args):
            pass
        def send(self,status,body,location=None):
            data=json.dumps(body).encode()
            self.send_response(status)
            self.send_header("Content-Type","application/json")
            self.send_header("Cache-Control","no-store")
            if location:
                self.send_header("Location",location)
            self.send_header("Content-Length",str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        def handle_request(self,method):
            try:
                auth=self.headers.get("Authorization","")
                owner=next((n for n,t in tokens.items() if hmac.compare_digest(auth.encode("utf-8"),("Bearer "+t).encode("utf-8"))),None)
                if owner is None:
                    raise Rejected(401,"valid bearer token required")
                parts=self.path.strip("/").split("/")
                if method=="POST" and parts==["v1","clusters"]:
                    try:
                        length=int(self.headers.get("Content-Length","0"))
                    except ValueError:
                        raise Rejected(400,"invalid length")
                    if not 1<=length<=4096:
                        raise Rejected(413,"body must be 1..4096 bytes")
                    try:
                        body=json.loads(self.rfile.read(length))
                    except (ValueError,UnicodeError):
                        raise Rejected(400,"invalid JSON")
                    if not isinstance(body,dict) or set(body)!={"plan"} or not isinstance(body["plan"],str):
                        raise Rejected(400,"only plan is accepted; owner comes from authentication")
                    op=request_create(path,owner,self.headers.get("Idempotency-Key",""),body["plan"])
                    self.send(202,op,"/v1/operations/"+op["id"])
                elif method=="GET" and parts==["v1","clusters"]:
                    self.send(200,inventory(path,owner))
                elif method=="GET" and len(parts)==3 and parts[:2] in (["v1","clusters"],["v1","operations"]):
                    self.send(200,owned(path,parts[1],owner,parts[2]))
                elif method=="DELETE" and len(parts)==3 and parts[:2]==["v1","clusters"]:
                    op=request_delete(path,owner,parts[2])
                    self.send(202,op,"/v1/operations/"+op["id"])
                else:
                    raise Rejected(404,"route not found")
            except Rejected as error:
                self.send(error.status,{"error":error.message})
        def do_GET(self): self.handle_request("GET")
        def do_POST(self): self.handle_request("POST")
        def do_DELETE(self): self.handle_request("DELETE")
    return ThreadingHTTPServer(address,Handler)

def serve(path):
    server=make_server(path)
    print("Educational API at http://127.0.0.1:18080; worker runs separately",flush=True)
    try:
        server.serve_forever()
    finally:
        server.server_close()

def selftest():
    def expect(status,fn):
        try:
            fn()
        except Rejected as e:
            assert e.status==status,(e.status,status)
        else:
            raise AssertionError("expected rejection "+str(status))
    with tempfile.TemporaryDirectory() as directory:
        db=Path(directory)/"model.sqlite"
        op=request_create(db,"alice","k1","small")
        assert request_create(db,"alice","k1","small")==op
        expect(409,lambda:request_create(db,"alice","k1","large"))
        expect(404,lambda:owned(db,"operations","bob",op["id"]))
        expect(404,lambda:request_delete(db,"bob",op["cluster_id"]))
        expect(75,lambda:reconcile(db,crash=True))
        assert inventory(db,"alice")["external_resources"]==[op["cluster_id"]]
        assert owned(db,"operations","alice",op["id"])["state"]=="Pending"
        reconcile(db)
        reconcile(db)
        assert inventory(db,"alice")["external_resources"]==[op["cluster_id"]]
        assert owned(db,"clusters","alice",op["cluster_id"])["state"]=="Ready"
        deletion=request_delete(db,"alice",op["cluster_id"])
        expect(409,lambda:request_create(db,"alice","k2","small"))
        expect(75,lambda:reconcile(db,crash=True))
        assert not inventory(db,"alice")["external_resources"]
        reconcile(db)
        assert request_delete(db,"alice",op["cluster_id"])==owned(db,"operations","alice",deletion["id"])
        def attempt(key):
            try: return request_create(db,"bob",key,"small")["id"]
            except Rejected as e: return e.status
        with ThreadPoolExecutor(max_workers=2) as pool:
            results=list(pool.map(attempt,["race1","race2"]))
        assert results.count(409)==1,results
        assert len(inventory(db,"bob")["clusters"])==1
    return {"result":"passed","checks":["repeat","payload conflict","object ownership","lost response",
      "restart","repeat reconcile","delete retry","quota held during delete","parallel quota"]}

def main():
    os.umask(0o077)
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--db",default="model.sqlite")
    sub=parser.add_subparsers(dest="command",required=True)
    for name in ("selftest","init","serve"):
        sub.add_parser(name)
    run=sub.add_parser("reconcile");run.add_argument("--crash",action="store_true")
    create=sub.add_parser("create")
    create.add_argument("owner",choices=["alice","bob"]);create.add_argument("key");create.add_argument("plan")
    listing=sub.add_parser("list");listing.add_argument("owner",choices=["alice","bob"])
    delete=sub.add_parser("delete")
    delete.add_argument("owner",choices=["alice","bob"]);delete.add_argument("cluster_id")
    args=parser.parse_args()
    try:
        if args.command=="selftest": result=selftest()
        elif args.command=="init": result=init_tokens(args.db)
        elif args.command=="serve": return serve(args.db)
        elif args.command=="create": result=request_create(args.db,args.owner,args.key,args.plan)
        elif args.command=="delete": result=request_delete(args.db,args.owner,args.cluster_id)
        elif args.command=="list": result=inventory(args.db,args.owner)
        else: reconcile(args.db,args.crash);result={"result":"reconciled"}
        print(json.dumps(result,indent=2))
    except Rejected as error:
        print(json.dumps({"error":error.message,"status":error.status}))
        raise SystemExit(error.status if error.status<256 else 1)

if __name__=="__main__":
    main()