crud.py

from datetime import datetime, timezone

from lnbits.db import Database, Filters, Page
from lnbits.helpers import urlsafe_short_hash

from .helpers import check_item_tags, split_tags
from .models import (
    CreateInventory,
    CreateInventoryUpdateLog,
    CreateItem,
    CreateManager,
    Inventory,
    InventoryLogFilters,
    InventoryUpdateLog,
    Item,
    ItemFilters,
    Manager,
    PublicInventory,
)

db = Database("ext_inventory")


async def get_inventories(user_id: str) -> Inventory | None:
    return await db.fetchone(
        """
        SELECT * FROM inventory.inventories
        WHERE user_id = :user_id
        """,
        {"user_id": user_id},
        model=Inventory,
    )


async def get_inventory(user_id: str, inventory_id: str) -> Inventory | None:
    return await db.fetchone(
        """
            SELECT * FROM inventory.inventories
            WHERE id = :inventory_id AND user_id = :user_id
        """,
        {"inventory_id": inventory_id, "user_id": user_id},
        Inventory,
    )


async def get_public_inventory(inventory_id: str) -> PublicInventory | None:
    return await db.fetchone(
        """
            SELECT * FROM inventory.inventories
            WHERE id = :inventory_id
        """,
        {"inventory_id": inventory_id},
        PublicInventory,
    )


async def create_inventory(user_id: str, data: CreateInventory) -> Inventory:
    inventory_id = urlsafe_short_hash()
    inventory = Inventory(
        id=inventory_id,
        user_id=user_id,
        **data.dict(),
    )
    await db.insert("inventory.inventories", inventory)
    return inventory


async def update_inventory(data: Inventory) -> Inventory:
    data.updated_at = datetime.now(timezone.utc)
    await db.update("inventory.inventories", data)
    return data


async def delete_inventory(user_id: str, inventory_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.inventories
        WHERE id = :inventory_id AND user_id = :user_id
        """,
        {"inventory_id": inventory_id, "user_id": user_id},
    )


async def get_inventory_items_paginated(
    inventory_id: str, filters: Filters[ItemFilters] | None = None
) -> Page[Item]:
    where = ["inventory_id = :inventory_id"]
    params = {"inventory_id": inventory_id}

    return await db.fetch_page(
        "SELECT * FROM inventory.items",
        where=where,
        values=params,
        filters=filters,
        model=Item,
    )


async def get_items_by_ids(inventory_id: str, item_ids: list[str]) -> list[Item]:
    if not item_ids:
        return []
    if isinstance(item_ids, str):
        item_ids = [item_ids]
    q = ",".join([f"'{item_id}'" for item_id in item_ids])
    items = await db.fetchall(
        f"""
        SELECT * FROM inventory.items
        WHERE inventory_id = :inventory_id AND id IN ({q})
        """,
        {"inventory_id": inventory_id},
        model=Item,
    )
    return items


async def get_item(item_id: str) -> Item | None:
    return await db.fetchone(
        """
        SELECT * FROM inventory.items
        WHERE id = :item_id
        """,
        {"item_id": item_id},
        model=Item,
    )


async def create_item(data: CreateItem) -> Item:
    item_id = urlsafe_short_hash()
    item = Item(
        id=item_id,
        **data.dict(),
    )
    if item.manager_id is not None and item.manager_id != "":
        item.is_active = False  # Items created by managers are inactive by default
    await db.insert("inventory.items", item)
    return item


async def update_item(data: Item) -> Item:
    data.updated_at = datetime.now(timezone.utc)
    await db.update("inventory.items", data)
    return data


async def delete_item(item_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.items
        WHERE id = :item_id
        """,
        {"item_id": item_id},
    )


async def delete_inventory_items(inventory_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.items
        WHERE inventory_id = :inventory_id
        """,
        {"inventory_id": inventory_id},
    )


async def create_manager(data: CreateManager) -> Manager:
    manager_id = urlsafe_short_hash()
    manager = Manager(
        id=manager_id,
        **data.dict(),
    )
    await db.insert("inventory.managers", manager)
    return manager


async def update_manager(data: Manager) -> Manager:
    data.updated_at = datetime.now(timezone.utc)
    await db.update("inventory.managers", data)
    return data


async def get_managers(inventory_id: str) -> list[Manager]:
    return await db.fetchall(
        """
        SELECT * FROM inventory.managers
        WHERE inventory_id = :inventory_id
        """,
        {"inventory_id": inventory_id},
        model=Manager,
    )


async def get_manager(manager_id: str) -> Manager | None:
    return await db.fetchone(
        """
        SELECT * FROM inventory.managers
        WHERE id = :manager_id
        """,
        {"manager_id": manager_id},
        model=Manager,
    )


async def delete_manager(manager_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.managers
        WHERE id = :manager_id
        """,
        {"manager_id": manager_id},
    )


async def delete_inventory_managers(inventory_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.managers
        WHERE inventory_id = :inventory_id
        """,
        {"inventory_id": inventory_id},
    )


async def get_inventory_items(inventory_id: str) -> list[Item]:
    return await db.fetchall(
        """
        SELECT * FROM inventory.items
        WHERE inventory_id = :inventory_id
        """,
        {"inventory_id": inventory_id},
        model=Item,
    )


def manager_can_access_item(manager: Manager, item: Item) -> bool:
    # No restriction when manager has no tag limitations
    if manager.tags is None:
        return True

    allowed_tags = split_tags(manager.tags)
    # Explicitly disallow when the manager has no allowed tags configured
    if not allowed_tags:
        return False

    item_tags = split_tags(item.tags)
    if not item_tags:
        return False

    return check_item_tags(allowed_tags, item_tags)


async def get_manager_items(inventory_id: str, manager: Manager) -> list[Item]:
    items = await get_inventory_items(inventory_id)
    return [item for item in items if manager_can_access_item(manager, item)]


## Log/Audit
async def create_inventory_update_log(
    data: CreateInventoryUpdateLog,
) -> None:
    await db.insert("inventory.audit_logs", data)


async def get_inventory_update_logs_paginated(
    inventory_id: str,
    filters: Filters[InventoryLogFilters] | None = None,
) -> Page[InventoryUpdateLog]:
    where = ["inventory_id = :inventory_id"]
    params = {"inventory_id": inventory_id}

    return await db.fetch_page(
        "SELECT * FROM inventory.audit_logs",
        where=where,
        values=params,
        filters=filters,
        model=InventoryUpdateLog,
    )


async def delete_inventory_update_logs(inventory_id: str) -> None:
    await db.execute(
        """
        DELETE FROM inventory.audit_logs
        WHERE inventory_id = :inventory_id
        """,
        {"inventory_id": inventory_id},
    )


async def check_idempotency(idempotency_key: str) -> bool:
    """
    Check if this request has already been processed.
    Returns True if already processed.
    """
    existing_log = await db.fetchone(
        """
        SELECT * FROM inventory.audit_logs
        WHERE idempotency_key = :idempotency_key
        """,
        {"idempotency_key": idempotency_key},
        model=InventoryUpdateLog,
    )
    return existing_log is not None