access/crud.py

import secrets

from lnbits.db import Database
from lnbits.helpers import urlsafe_short_hash

from .models import AccessLog, CreateDoor, CreateGrant, Door, Grant

db = Database("ext_access")


# ── Doors ──────────────────────────────────────────────────────────────────
async def create_door(wallet_id: str, data: CreateDoor) -> Door:
    door_id = urlsafe_short_hash()
    controller_token = data.controller_token or secrets.token_urlsafe(24)
    await db.execute(
        """
        INSERT INTO access.doors (
            id, wallet, name, controller_token, ha_webhook_url,
            boltcards_base_url, unlock_timeout_ms, enabled
        )
        VALUES (
            :id, :wallet, :name, :controller_token, :ha_webhook_url,
            :boltcards_base_url, :unlock_timeout_ms, :enabled
        )
        """,
        {
            "id": door_id,
            "wallet": wallet_id,
            "name": data.name,
            "controller_token": controller_token,
            "ha_webhook_url": data.ha_webhook_url,
            "boltcards_base_url": data.boltcards_base_url,
            "unlock_timeout_ms": data.unlock_timeout_ms,
            "enabled": data.enabled,
        },
    )
    door = await get_door(door_id)
    assert door, "Newly created door couldn't be retrieved"
    return door


async def update_door(door: Door) -> Door:
    await db.update("access.doors", door)
    return door


async def get_door(door_id: str) -> Door | None:
    return await db.fetchone(
        "SELECT * FROM access.doors WHERE id = :id", {"id": door_id}, Door
    )


async def get_door_by_id_or_name(ident: str) -> Door | None:
    """The reader's `doorId` may be the door id or its human name."""
    return await db.fetchone(
        "SELECT * FROM access.doors WHERE id = :ident OR name = :ident",
        {"ident": ident},
        Door,
    )


async def get_doors(wallet_ids: list[str]) -> list[Door]:
    if not wallet_ids:
        return []
    q = ",".join(f"'{w}'" for w in wallet_ids)
    return await db.fetchall(
        f"SELECT * FROM access.doors WHERE wallet IN ({q}) ORDER BY name", model=Door
    )


async def delete_door(door_id: str) -> None:
    await db.execute("DELETE FROM access.doors WHERE id = :id", {"id": door_id})
    await db.execute(
        "DELETE FROM access.grants WHERE door_id = :id", {"id": door_id}
    )


# ── Grants ─────────────────────────────────────────────────────────────────
async def create_grant(data: CreateGrant) -> Grant:
    grant_id = urlsafe_short_hash()
    await db.execute(
        """
        INSERT INTO access.grants (id, door_id, external_id, label, enabled, expires_at)
        VALUES (:id, :door_id, :external_id, :label, :enabled, :expires_at)
        """,
        {
            "id": grant_id,
            "door_id": data.door_id,
            "external_id": data.external_id.lower(),
            "label": data.label,
            "enabled": data.enabled,
            "expires_at": data.expires_at,
        },
    )
    grant = await get_grant(grant_id)
    assert grant, "Newly created grant couldn't be retrieved"
    return grant


async def get_grant(grant_id: str) -> Grant | None:
    return await db.fetchone(
        "SELECT * FROM access.grants WHERE id = :id", {"id": grant_id}, Grant
    )


async def get_grants(door_ids: list[str]) -> list[Grant]:
    if not door_ids:
        return []
    q = ",".join(f"'{d}'" for d in door_ids)
    return await db.fetchall(
        f"SELECT * FROM access.grants WHERE door_id IN ({q}) ORDER BY time DESC",
        model=Grant,
    )


async def get_active_grant(door_id: str, external_id: str) -> Grant | None:
    """The permission decision: an enabled, unexpired grant for this card+door."""
    return await db.fetchone(
        """
        SELECT * FROM access.grants
        WHERE door_id = :door_id AND external_id = :external_id AND enabled = true
        """,
        {"door_id": door_id, "external_id": external_id.lower()},
        Grant,
    )


async def delete_grant(grant_id: str) -> None:
    await db.execute("DELETE FROM access.grants WHERE id = :id", {"id": grant_id})


# ── Access log ─────────────────────────────────────────────────────────────
async def record_log(
    door_id: str, external_id: str, decision: str, reason: str, ip: str = ""
) -> None:
    await db.execute(
        """
        INSERT INTO access.logs (id, door_id, external_id, decision, reason, ip)
        VALUES (:id, :door_id, :external_id, :decision, :reason, :ip)
        """,
        {
            "id": urlsafe_short_hash(),
            "door_id": door_id,
            "external_id": external_id,
            "decision": decision,
            "reason": reason,
            "ip": ip,
        },
    )


async def get_logs(door_ids: list[str], limit: int = 200) -> list[AccessLog]:
    if not door_ids:
        return []
    q = ",".join(f"'{d}'" for d in door_ids)
    return await db.fetchall(
        f"SELECT * FROM access.logs WHERE door_id IN ({q}) "
        f"ORDER BY time DESC LIMIT {int(limit)}",
        model=AccessLog,
    )