update-server/app/repositories/license_repository.py

import db


def insert_license(
    license_id: str,
    license_key_hash: str,
    license_key_cipher: str,
    customer_name: str,
    app_id: str,
    channel: str,
    max_devices: int,
    valid_until: str,
):
    conn = db.get_conn()
    try:
        channel_row = conn.execute("SELECT * FROM channels WHERE app_id=? AND channel_code=?", (app_id, channel)).fetchone()
        if not channel_row:
            return "channel_not_found"
        if not channel_row["enabled"]:
            return "channel_disabled"
        conn.execute(
            """INSERT INTO licenses
            (license_id,license_key_hash,license_key_cipher,customer_name,app_id,channel_code,max_devices,valid_until,status)
            VALUES(?,?,?,?,?,?,?,?,'active')""",
            (license_id, license_key_hash, license_key_cipher, customer_name, app_id, channel, max_devices, valid_until),
        )
        conn.commit()
        return "ok"
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def list_licenses(app_id: str = "", include_deleted: bool = False) -> list:
    deleted_filter = "" if include_deleted else " AND status<>'deleted'"
    conn = db.get_conn()
    if app_id:
        rows = conn.execute(
            f"""SELECT l.*,(SELECT COUNT(*) FROM license_devices d WHERE d.license_id=l.license_id) used_devices
            FROM licenses l WHERE app_id=?{deleted_filter} ORDER BY created_at DESC""",
            (app_id,),
        ).fetchall()
    else:
        rows = conn.execute(
            f"""SELECT l.*,(SELECT COUNT(*) FROM license_devices d WHERE d.license_id=l.license_id) used_devices
            FROM licenses l WHERE 1=1{deleted_filter} ORDER BY created_at DESC"""
        ).fetchall()
    conn.close()
    return rows


def get_license_by_hash_with_usage(license_key_hash: str):
    conn = db.get_conn()
    row = conn.execute(
        """SELECT l.*,(SELECT COUNT(*) FROM license_devices d WHERE d.license_id=l.license_id) used_devices
        FROM licenses l WHERE l.license_key_hash=? AND l.status<>'deleted'""",
        (license_key_hash,),
    ).fetchone()
    conn.close()
    return row


def set_license_status(license_id: str, status: str) -> int:
    conn = db.get_conn()
    cur = conn.execute(
        "UPDATE licenses SET status=?,updated_at=datetime('now') WHERE license_id=? AND status<>'deleted'",
        (status, license_id),
    )
    conn.commit()
    updated = cur.rowcount
    conn.close()
    return updated


def soft_delete_license(license_id: str) -> int:
    conn = db.get_conn()
    cur = conn.execute(
        "UPDATE licenses SET status='deleted',updated_at=datetime('now') WHERE license_id=? AND status<>'deleted'",
        (license_id,),
    )
    conn.commit()
    updated = cur.rowcount
    conn.close()
    return updated