Ledger events
=============

Events are named, user-defined operations on a ledger.  Each event
compiles into a single PostgreSQL function (``LANGUAGE sql``) that
inserts rows into the ledger view and returns the inserted rows
via ``RETURNING *``.

Import from the ledger extension::

    from codegen_database.ext.ledger import (
        LedgerEvent,
        construct_ledger_balances_query,
    )


Reconciliation (diff mode)
--------------------------

The primary pattern is **reconciliation**: you have a transactional
object (invoices, shipments, work orders) with line items, and you
want the ledger to reflect the current state of those line items.
The system handles the arithmetic — it diffs the desired state against
existing balances and inserts only the correcting deltas.

This makes the operation **idempotent**: calling the same event twice
with the same input produces zero new rows the second time.  Amending
the source and re-calling inserts exactly the correcting entries.

**Example** — revenue recognition from invoice line items::

    from sqlalchemy import Integer, String, func, select
    from sqlalchemy.dialects.postgresql import ARRAY

    from codegen_database.ext.ledger import (
        LedgerEvent,
        construct_ledger_balances_query,
    )

    # Assume `invoice_lines` is a SQLAlchemy Table with columns:
    #   id, invoice_id, account, amount

    recognize = LedgerEvent(
        name="recognize",
        input=lambda p: select(
            func.unnest(p("invoice_ids", ARRAY(Integer)))
            .label("invoice_id"),
        ),
        desired=lambda pginput: select(
            invoice_lines.c.invoice_id,
            invoice_lines.c.account,
            invoice_lines.c.amount.label("value"),
        ).where(
            invoice_lines.c.invoice_id.in_(
                select(pginput.c.invoice_id)
            )
        ),
        existing=construct_ledger_balances_query("invoice_id", "account"),
        diff_keys=["invoice_id", "account"],
    )

The generated SQL uses CTEs: ``input`` (the function parameters),
``desired`` (the target state from the source table), ``existing``
(negated current balances), and ``deltas`` (the union of desired +
existing, filtered to non-zero values).  This is a single
``INSERT ... RETURNING *`` statement.

Key parameters:

``input``
    Lambda ``(p) -> Select`` that builds the input CTE.  ``p`` is a
    :class:`~codegen_database.ext.ledger.events.ParamCollector` — each call to
    ``p("name", Type)`` records a function parameter and returns a
    ``literal_column("p_name")`` reference.

``desired``
    Lambda ``(pginput) -> Select`` that builds the desired-state CTE.
    ``pginput`` is a synthetic table reference to the ``input`` CTE.
    Typically joins against a source table (e.g. ``invoice_lines``)
    to produce the ledger entries.

``existing``
    Lambda ``(table, desired, pginput) -> Select`` that returns the
    negated current balances.  ``pginput`` is the same input-CTE
    reference passed to ``desired``, so function-param values are
    reached as ``pginput.c.<name>`` -- no ``p_`` prefix.  Use
    :func:`~codegen_database.ext.ledger.events.construct_ledger_balances_query`
    or
    :func:`~codegen_database.ext.ledger.events.construct_ledger_scoped_balances_query`
    for the common patterns.

``diff_keys``
    Column names used for grouping.  Required when ``desired`` is set.

The ``construct_ledger_balances_query`` helper produces an ``existing`` callable that
groups ``SUM(value) * -1`` by the specified keys, filtered to only
groups present in the desired CTE.


Owner-scoped reconciliation
~~~~~~~~~~~~~~~~~~~~~~~~~~~

``construct_ledger_balances_query`` filters ``existing`` to rows
whose diff-key tuple appears in ``desired``.  That works when every
key the event might want to unwind is guaranteed to also be in
``desired`` -- e.g. invoice recognition where each line item always
maps to exactly one ``(invoice_id, account)`` pair.

It does **not** work when a state transition can *retire* a key.
If an event posts ``in_transit`` while a lot is in flight, then
re-posts under ``active`` once delivered, the ``in_transit`` row no
longer appears in ``desired`` after delivery -- so the default
helper can't see it, and the unwinding entry never gets emitted.

For that pattern, use
:func:`~codegen_database.ext.ledger.events.construct_ledger_scoped_balances_query`,
which filters ``existing`` by an arbitrary predicate over the
ledger root instead::

    from codegen_database.ext.ledger import construct_ledger_scoped_balances_query

    def owned_by_input_po(ledger, pginput):
        return ledger.c.lot_id.in_(
            select(PurchaseLot.id)
            .join(LineItem, LineItem.id == PurchaseLot.line_item_id)
            .where(
                LineItem.purchase_order_id
                == pginput.c.purchase_order_id,
            )
        )

    post_po = LedgerEvent(
        name="post_po",
        input=...,
        desired=...,
        existing=construct_ledger_scoped_balances_query(
            "lot_id", "class_id", "account", "direction",
            where=owned_by_input_po,
        ),
        diff_keys=["lot_id", "class_id", "account", "direction"],
    )

The ``where`` predicate receives the ledger root table and the
``pginput`` CTE reference (same as ``desired``), so function
parameters are reached as ``pginput.c.<name>`` -- no ``p_``
prefix.  Reach the owner directly via a column when it's
denormalized on the ledger, or via a subquery / join when it's
not.


Usage
~~~~~

.. code-block:: sql

   -- Recognize revenue for two invoices:
   SELECT * FROM ops.ops_revenue_recognize(
       p_invoice_ids => ARRAY[1001, 1002]
   );

   -- Call again — idempotent, returns 0 rows:
   SELECT * FROM ops.ops_revenue_recognize(
       p_invoice_ids => ARRAY[1001, 1002]
   );

   -- Amend invoice 1001's line items in the source table,
   -- then re-reconcile — only correcting deltas are inserted:
   SELECT * FROM ops.ops_revenue_recognize(
       p_invoice_ids => ARRAY[1001]
   );


Simple event (input only)
-------------------------

A simple event is a special case — it provides only ``input`` and
inserts directly, without diffing.  This is what happens when you
omit ``desired``, ``existing``, and ``diff_keys``: there are no
existing entries to diff against, so every call inserts new rows.

Use this for fire-and-forget deltas where idempotency is not needed
(stock adjustments, manual corrections, one-off charges)::

    adjust = LedgerEvent(
        name="adjust",
        input=lambda p: select(
            p("warehouse", String).label("warehouse"),
            p("sku", String).label("sku"),
            p("value", Integer).label("value"),
            p("reason", String).label("reason"),
        ),
    )

.. tip::

   If your simple event maps one-to-one with an external identifier
   (e.g. an ``invoice_id``), prefer diff mode instead.  Diffing over
   that identifier makes the operation idempotent — calling the same
   event twice with the same input produces zero new rows the second
   time.


Factory integration
-------------------

The recommended style is **declarative**: define the ledger model as a
:class:`~codegen_database.declarative.CodegenDatabaseBase` subclass, then attach events
with :class:`~codegen_database.declarative.CodegenDatabaseFunctionMixin` and
:func:`~codegen_database.ext.ledger.functions.ledger_event_function`::

    from sqlalchemy import Column, Integer, String
    from codegen_database import CodegenDatabaseFunctionMixin
    from codegen_database.ext.ledger import (
        LedgerEvent,
        construct_ledger_balances_query,
        ledger_event_function,
    )
    from codegen_database.factory import CodegenDatabaseLedger


    class Revenue(Base):
        __tablename__ = "revenue"
        __table_args__ = {"schema": "ops"}
        __factory__ = CodegenDatabaseLedger

        invoice_id = Column(Integer, nullable=False)
        account = Column(String, nullable=False)


    class RecognizeRevenue(CodegenDatabaseFunctionMixin, Base):
        __table_args__ = {"schema": "ops"}
        __funcspec__ = ledger_event_function(
            Revenue,
            LedgerEvent(
                name="recognize",
                input=lambda p: select(
                    func.unnest(p("invoice_ids", ARRAY(Integer)))
                    .label("invoice_id"),
                ),
                desired=lambda pginput: select(
                    invoice_lines.c.invoice_id,
                    invoice_lines.c.account,
                    invoice_lines.c.amount.label("value"),
                ).where(
                    invoice_lines.c.invoice_id.in_(
                        select(pginput.c.invoice_id)
                    )
                ),
                existing=construct_ledger_balances_query("invoice_id", "account"),
                diff_keys=["invoice_id", "account"],
            ),
        )


Imperative alternative
~~~~~~~~~~~~~~~~~~~~~~

If you build your schema programmatically, call
:class:`~codegen_database.factory.ledger.CodegenDatabaseLedger` and
:class:`~codegen_database.functions.CodegenDatabaseFunction` directly.  Because
:class:`~codegen_database.functions.CodegenDatabaseFunctionSpec` does not include a schema,
expand the spec and pass ``schema`` explicitly::

    from codegen_database.factory import CodegenDatabaseLedger
    from codegen_database.functions import CodegenDatabaseFunction
    from codegen_database.ext.ledger import ledger_event_function

    revenue = CodegenDatabaseLedger(
        tablename="revenue",
        schemaname="ops",
        metadata=metadata,
        schema_items=[
            Column("invoice_id", Integer, nullable=False),
            Column("account", String, nullable=False),
        ],
    )

    spec = ledger_event_function(revenue, recognize)
    CodegenDatabaseFunction(
        name=spec["name"],
        schema=revenue.ctx.schemaname,
        definition=spec["definition"],
        metadata=metadata,
        returns=spec["returns"],
        language=spec["language"],
        parameters=spec["parameters"],
        security=spec["security"],
    )


Naming convention
~~~~~~~~~~~~~~~~~

Generated function names follow the pattern
``{schema}_{table}_{name}`` and are registered in the ledger's schema:

=========================  =====================================
Function                   Example
=========================  =====================================
``{table}_{name}``         ``ops_revenue_recognize``
``{table}_{name}``         ``ops_inventory_adjust``
=========================  =====================================

Event names must be unique within a ledger.


Validation
----------

Event configuration is validated when
:func:`~codegen_database.ext.ledger.functions.ledger_event_function` is called:

- ``diff_keys`` is required when ``desired`` is set.
- ``existing`` requires ``desired`` to be set.
- Every key in ``diff_keys`` must name a column on the ledger table.

Violations raise :class:`~codegen_database.errors.CodegenDatabaseValidationError`.