docs(billing): TDD plan for NexaCloud invoice ledger (ingest -> account.move, posted+reconciled+HST)

2026-05-27 16:44:21 -04:00
parent bf6ee2bb2c
commit f6518b4d7e
1 changed files with 637 additions and 0 deletions
--- a/docs/superpowers/plans/2026-05-27-nexacloud-invoice-ledger.md
+++ b/docs/superpowers/plans/2026-05-27-nexacloud-invoice-ledger.md
@@ -0,0 +1,637 @@
+# NexaCloud → Odoo Invoice Ledger — Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: superpowers:executing-plans. Steps use checkbox (`- [ ]`) syntax.
+
+**Goal:** Ingest NexaCloud's real (Stripe-billed) invoices into Odoo as posted `account.move` customer invoices with reconciled payments + HST, so Odoo is the accounting system of record — all history + ongoing, revenue split by service family, draft-first on the live books.
+
+**Architecture:** A new ingester in `fusion_centralize_billing` mirroring the importer's read/write split: `_read_nexacloud_invoices` (read-only psycopg2 via the existing DSN) → `_ingest_invoices` (pure Odoo: create `account.move` drafts idempotently, map lines to per-family income accounts, derive tax, reconcile Stripe payments) → `_post_ingested` (bulk-post after review). Reuses the `account.link` partner mapping. Native Odoo accounting does the rest.
+
+**Tech Stack:** Odoo 19 Enterprise, `account_accountant`, `psycopg2`. Tests: `TransactionCase` on odoo-trial (`bash scripts/fcb_test_on_trial.sh`, pass = `FCB_EXIT=0`).
+
+**Spec:** `docs/superpowers/specs/2026-05-27-nexacloud-invoice-ledger-design.md`
+
+---
+
+## Conventions
+- **Never code accounting internals from memory** (CLAUDE rule #1). Reference confirmed on trial: `account.move` has `invoice_line_ids`/`invoice_date`/`action_post`; `account.payment.register` exists; `account_type='income'`/`'asset_receivable'` valid; sale taxes are Canadian (find HST 13% by `amount=13` / name). Where a step says "read reference", confirm before relying on it.
+- **Models, not UI:** logic in model methods; the wizard only calls them. Testable under `TransactionCase`.
+- **New fields on native models:** `x_fc_*`. Declarative `models.Constraint` only.
+- Tests run on **odoo-trial** (`bash scripts/fcb_test_on_trial.sh`, full suite, ~1–2 min). Register each new `tests/test_*.py` in `tests/__init__.py` in the same task.
+
+## File structure
+```
+fusion_centralize_billing/
+  models/
+    account_move.py            # NEW: account.move inherit (x_fc_nexacloud_invoice_id, x_fc_stripe_invoice_id)
+    __init__.py                # + account_move
+  wizards/
+    invoice_ledger.py          # NEW: the ingester (read + ingest + post + family/tax/payment helpers)
+    __init__.py                # + invoice_ledger
+  views/
+    invoice_ledger_views.xml   # NEW: wizard form + action + menu + cron
+  security/ir.model.access.csv # + ledger wizard ACL
+  __manifest__.py              # + views/invoice_ledger_views.xml
+  tests/
+    test_invoice_ledger.py     # NEW
+    __init__.py                # + test_invoice_ledger
+```
+
+---
+
+## Task 1: Scaffold — account.move fields + ledger wizard skeleton
+
+**Files:** create `models/account_move.py`, `wizards/invoice_ledger.py`, `views/invoice_ledger_views.xml`; modify `models/__init__.py`, `wizards/__init__.py`, `security/ir.model.access.csv`, `__manifest__.py`.
+
+- [ ] **Step 1: account.move inherit** — `models/account_move.py`:
+```python
+# -*- coding: utf-8 -*-
+# Copyright 2026 Nexa Systems Inc.
+# License OPL-1
+from odoo import fields, models
+
+
+class AccountMove(models.Model):
+    _inherit = "account.move"
+
+    x_fc_nexacloud_invoice_id = fields.Char(
+        index=True, copy=False, help="Source NexaCloud invoice id — ledger idempotency key.")
+    x_fc_stripe_invoice_id = fields.Char(index=True, copy=False)
+
+    _fc_nc_invoice_uniq = models.Constraint(
+        "unique(x_fc_nexacloud_invoice_id)",
+        "One Odoo invoice per NexaCloud invoice id.")
+```
+Add `from . import account_move` to `models/__init__.py`.
+
+- [ ] **Step 2: ledger wizard skeleton** — `wizards/invoice_ledger.py`:
+```python
+# -*- coding: utf-8 -*-
+# Copyright 2026 Nexa Systems Inc.
+# License OPL-1
+import json
+import logging
+
+from odoo import api, fields, models
+from odoo.exceptions import UserError
+
+_logger = logging.getLogger(__name__)
+
+
+class FusionBillingInvoiceLedgerWizard(models.TransientModel):
+    _name = "fusion.billing.invoice.ledger.wizard"
+    _description = "Fusion Billing — NexaCloud Invoice Ledger Ingester"
+
+    dry_run = fields.Boolean(default=True)
+    auto_post = fields.Boolean(
+        default=False, help="Post invoices immediately (else leave draft for review).")
+    result_summary = fields.Text(readonly=True)
+
+    def _ingest_invoices(self, data, post=False):
+        return {"created": 0, "updated": 0, "posted": 0, "skipped": [], "failed": [], "by_family": {}}
+```
+Add `from . import invoice_ledger` to `wizards/__init__.py`.
+
+- [ ] **Step 3: view + action + menu** — `views/invoice_ledger_views.xml`:
+```xml
+<?xml version="1.0" encoding="utf-8"?>
+<odoo>
+    <record id="view_fc_invoice_ledger_wizard_form" model="ir.ui.view">
+        <field name="name">fusion.billing.invoice.ledger.wizard.form</field>
+        <field name="model">fusion.billing.invoice.ledger.wizard</field>
+        <field name="arch" type="xml">
+            <form string="Ingest NexaCloud Invoices">
+                <group>
+                    <field name="dry_run"/>
+                    <field name="auto_post"/>
+                </group>
+                <group string="Result" invisible="not result_summary">
+                    <field name="result_summary" nolabel="1" widget="text"/>
+                </group>
+                <footer>
+                    <button name="action_run" type="object" string="Run" class="btn-primary"/>
+                    <button string="Close" class="btn-secondary" special="cancel"/>
+                </footer>
+            </form>
+        </field>
+    </record>
+    <record id="action_fc_invoice_ledger_wizard" model="ir.actions.act_window">
+        <field name="name">Ingest NexaCloud Invoices</field>
+        <field name="res_model">fusion.billing.invoice.ledger.wizard</field>
+        <field name="view_mode">form</field>
+        <field name="target">new</field>
+    </record>
+    <menuitem id="menu_fc_invoice_ledger" name="Ingest NexaCloud Invoices"
+              parent="menu_fusion_billing_root"
+              action="action_fc_invoice_ledger_wizard" sequence="20"
+              groups="base.group_system"/>
+</odoo>
+```
+
+- [ ] **Step 4: security + manifest** — append to `security/ir.model.access.csv`:
+```
+access_fc_invoice_ledger_wizard,fusion.billing.invoice.ledger.wizard,model_fusion_billing_invoice_ledger_wizard,base.group_system,1,1,1,1
+```
+Add `"views/invoice_ledger_views.xml"` to `__manifest__.py` `data`.
+
+- [ ] **Step 5: verify upgrade** — `bash scripts/fcb_test_on_trial.sh` → `FCB_EXIT=0` (existing tests pass; new model/fields/view load).
+
+- [ ] **Step 6: commit** — `feat(billing): invoice-ledger scaffold (account.move x_fc fields + wizard)`
+
+---
+
+## Task 2: Service-family classification + income account
+
+**Files:** modify `wizards/invoice_ledger.py`; create `tests/test_invoice_ledger.py` (+ register in `tests/__init__.py`).
+
+- [ ] **Step 1: failing test** — `tests/test_invoice_ledger.py`:
+```python
+# -*- coding: utf-8 -*-
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('post_install', '-at_install')
+class TestLedgerFamily(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
+
+    def test_family_classification(self):
+        f = self.W._fc_family_for
+        self.assertEqual(f('Odoo ERP Hosting (2026-05-01 to 2026-06-01)'), 'hosting')
+        self.assertEqual(f('WordPress Website Hosting - Managed (at $50.00 / month)'), 'hosting')
+        self.assertEqual(f('Managed Odoo - Standard (at $49.99 / month)'), 'managed')
+        self.assertEqual(f('Daily Backup Protection'), 'addons')
+        self.assertEqual(f('Remaining time on Daily Backup Protection after 27 May 2026'), 'addons')
+        self.assertEqual(f('Something Unmapped'), 'other')
+
+    def test_income_account_per_family_distinct(self):
+        a_host = self.W._fc_income_account('hosting')
+        a_add = self.W._fc_income_account('addons')
+        self.assertEqual(a_host.account_type, 'income')
+        self.assertNotEqual(a_host, a_add)               # split by family
+        self.assertEqual(self.W._fc_income_account('hosting'), a_host)  # idempotent
+```
+Append `from . import test_invoice_ledger` to `tests/__init__.py`.
+
+- [ ] **Step 2: run** → FAIL (`_fc_family_for` missing).
+
+- [ ] **Step 3: implement** — in `wizards/invoice_ledger.py`:
+```python
+    _FAMILY_KEYWORDS = [
+        ('hosting', ['odoo erp hosting', 'wordpress website hosting']),
+        ('managed', ['managed']),
+        ('addons', ['daily backup', 'whatsapp', 'forms builder', 'white label']),
+    ]
+
+    @api.model
+    def _fc_family_for(self, description):
+        import re
+        d = (description or '').lower()
+        m = re.match(r'remaining time on (.+?)(?: after| from |\s*\()', d)
+        if m:
+            d = m.group(1)  # classify proration by the prorated item
+        for fam, kws in self._FAMILY_KEYWORDS:
+            if any(k in d for k in kws):
+                return fam
+        return 'other'
+
+    @api.model
+    def _fc_income_account(self, family):
+        Account = self.env['account.account']
+        code = 'NCR-' + family.upper()[:6]
+        acc = Account.search([('code', '=', code)], limit=1)
+        if not acc:
+            acc = Account.create({
+                'code': code, 'name': 'NexaCloud %s Revenue' % family.title(),
+                'account_type': 'income'})
+        return acc
+```
+
+- [ ] **Step 4: run** → PASS. (If `account.account.create` needs more required fields on this build, read `account_account.py` on trial and add them — don't weaken the test.)
+
+- [ ] **Step 5: commit** — `feat(billing): ledger service-family classification + per-family income accounts`
+
+---
+
+## Task 3: Tax derivation (match NexaCloud's invoice.tax)
+
+**Files:** modify `wizards/invoice_ledger.py`, `tests/test_invoice_ledger.py`.
+
+- [ ] **Step 1: failing test** (append):
+```python
+@tagged('post_install', '-at_install')
+class TestLedgerTax(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
+
+    def test_tax_for_13pct_is_a_13_percent_sale_tax(self):
+        tax = self.W._fc_tax_for(100.0, 13.0)
+        self.assertTrue(tax, "expected an HST/13% sale tax on the Canadian COA")
+        self.assertEqual(tax.type_tax_use, 'sale')
+        # the chosen tax computes 13.00 on 100.00
+        res = tax.compute_all(100.0)
+        self.assertAlmostEqual(res['total_included'] - res['total_excluded'], 13.0, places=2)
+
+    def test_tax_for_zero_is_zero_or_empty(self):
+        tax = self.W._fc_tax_for(100.0, 0.0)
+        if tax:
+            res = tax.compute_all(100.0)
+            self.assertAlmostEqual(res['total_included'] - res['total_excluded'], 0.0, places=2)
+```
+
+- [ ] **Step 2: run** → FAIL.
+
+- [ ] **Step 3: implement**:
+```python
+    @api.model
+    def _fc_tax_for(self, subtotal, tax_amount):
+        """Map a NexaCloud invoice's (subtotal, tax_amount) to the Odoo sale tax whose
+        computed tax equals it. Picks by effective percent; falls back to a 0% sale tax."""
+        Tax = self.env['account.tax']
+        sub = float(subtotal or 0.0)
+        tax_amt = float(tax_amount or 0.0)
+        if sub <= 0 or tax_amt <= 0:
+            return Tax.search([('type_tax_use', '=', 'sale'), ('amount', '=', 0.0)], limit=1)
+        rate = round(100.0 * tax_amt / sub)
+        tax = Tax.search([('type_tax_use', '=', 'sale'), ('amount_type', '=', 'percent'),
+                          ('amount', '=', float(rate))], limit=1)
+        if not tax:
+            tax = Tax.search([('type_tax_use', '=', 'sale'), ('name', 'ilike', '%s' % rate)], limit=1)
+        return tax
+```
+
+- [ ] **Step 4: run** → PASS. (Read reference if no 13% sale tax exists: `docker exec odoo-trial-app ... grep -i hst` the l10n_ca data; on nexamain confirm the HST 13% record from `nexa_coa_setup`.)
+
+- [ ] **Step 5: commit** — `feat(billing): ledger tax derivation matching source invoice tax`
+
+---
+
+## Task 4: Ingest invoices → draft account.move (idempotent)
+
+**Read reference first:**
+```bash
+ssh pve-worker1 "qm guest exec 316 -- bash -lc 'docker exec odoo-trial-app bash -lc \"grep -nE \\\"def action_post|invoice_line_ids|move_type\\\" /mnt/enterprise-addons/account_accountant/../account/models/account_move.py | head\"'"
+```
+Confirm `account.move.create({'move_type':'out_invoice','partner_id':..,'invoice_line_ids':[(0,0,{'name','quantity','price_unit','account_id','tax_ids'})]})` and `move.amount_untaxed/amount_tax/amount_total`.
+
+**Files:** modify `wizards/invoice_ledger.py`, `tests/test_invoice_ledger.py`.
+
+- [ ] **Step 1: failing test** (append) — uses a fixture invoice dict shaped like `_read_nexacloud_invoices` output:
+```python
+def _inv_fixture():
+    return [{
+        'id': 'inv-1', 'stripe_invoice_id': 'in_test1', 'invoice_number': 'NEX-0001',
+        'user_external_id': 'u-1', 'partner_name': 'Acme', 'partner_email': 'ar@acme.test',
+        'invoice_date': '2026-05-01', 'currency': 'CAD', 'status': 'open',
+        'subtotal': 100.0, 'tax': 13.0, 'amount_paid': 0.0, 'paid_at': None,
+        'items': [{'description': 'Odoo ERP Hosting (2026-05-01 to 2026-06-01)',
+                   'quantity': 1.0, 'unit_price': 100.0, 'amount': 100.0}],
+    }]
+
+
+@tagged('post_install', '-at_install')
+class TestLedgerIngest(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.W = self.env['fusion.billing.invoice.ledger.wizard'].sudo()
+        self.svc = self.env['fusion.billing.service'].sudo().create(
+            {'name': 'NexaCloud', 'code': 'nexacloud'})
+
+    def test_ingest_creates_draft_invoice_with_right_totals(self):
+        self.W._ingest_invoices(_inv_fixture(), post=False)
+        mv = self.env['account.move'].search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
+        self.assertEqual(len(mv), 1)
+        self.assertEqual(mv.move_type, 'out_invoice')
+        self.assertEqual(mv.state, 'draft')
+        self.assertAlmostEqual(mv.amount_untaxed, 100.0, places=2)
+        self.assertAlmostEqual(mv.amount_tax, 13.0, places=2)      # equals source tax
+        self.assertAlmostEqual(mv.amount_total, 113.0, places=2)
+        self.assertEqual(mv.partner_id.email, 'ar@acme.test')
+        line = mv.invoice_line_ids
+        self.assertEqual(line.account_id, self.W._fc_income_account('hosting'))
+
+    def test_ingest_is_idempotent(self):
+        self.W._ingest_invoices(_inv_fixture(), post=False)
+        self.W._ingest_invoices(_inv_fixture(), post=False)
+        self.assertEqual(self.env['account.move'].search_count(
+            [('x_fc_nexacloud_invoice_id', '=', 'inv-1')]), 1)
+```
+
+- [ ] **Step 2: run** → FAIL.
+
+- [ ] **Step 3: implement** the partner resolver + `_ingest_invoices`:
+```python
+    @api.model
+    def _fc_partner_for(self, inv):
+        """Resolve the unified partner for an invoice via the nexacloud account.link
+        (by user_external_id); create partner+link if missing (covers NULL-subscription
+        invoices, which still carry a user)."""
+        service = self.env['fusion.billing.service'].search([('code', '=', 'nexacloud')], limit=1)
+        link = self.env['fusion.billing.account.link']._resolve_or_create_partner(
+            service, str(inv.get('user_external_id')),
+            name=inv.get('partner_name'), email=inv.get('partner_email'))
+        return link.partner_id
+
+    @api.model
+    def _ingest_invoices(self, data, post=False):
+        Move = self.env['account.move']
+        cad = self.env.ref('base.CAD', raise_if_not_found=False) or self.env.company.currency_id
+        summary = {'created': 0, 'updated': 0, 'posted': 0, 'skipped': [], 'failed': [], 'by_family': {}}
+        for inv in data:
+            nc_id = str(inv.get('id') or '')
+            try:
+                with self.env.cr.savepoint():
+                    existing = Move.search([('x_fc_nexacloud_invoice_id', '=', nc_id)], limit=1)
+                    if existing:
+                        if existing.state != 'draft':
+                            summary['skipped'].append({'id': nc_id, 'reason': 'already posted'})
+                            continue
+                        existing.invoice_line_ids.unlink()  # draft: replace lines
+                        move = existing
+                    else:
+                        move = Move.create({
+                            'move_type': 'out_invoice',
+                            'partner_id': self._fc_partner_for(inv).id,
+                            'invoice_date': inv.get('invoice_date'),
+                            'ref': inv.get('invoice_number'),
+                            'currency_id': cad.id,
+                            'x_fc_nexacloud_invoice_id': nc_id,
+                            'x_fc_stripe_invoice_id': inv.get('stripe_invoice_id'),
+                        })
+                    tax = self._fc_tax_for(inv.get('subtotal'), inv.get('tax'))
+                    line_vals = []
+                    for it in inv.get('items', []):
+                        fam = self._fc_family_for(it.get('description'))
+                        summary['by_family'][fam] = round(
+                            summary['by_family'].get(fam, 0.0) + float(it.get('amount') or 0.0), 2)
+                        line_vals.append((0, 0, {
+                            'name': it.get('description') or 'NexaCloud',
+                            'quantity': float(it.get('quantity') or 1.0),
+                            'price_unit': float(it.get('unit_price') or it.get('amount') or 0.0),
+                            'account_id': self._fc_income_account(fam).id,
+                            'tax_ids': [(6, 0, tax.ids)] if tax else [(5, 0, 0)],
+                        }))
+                    move.write({'invoice_line_ids': line_vals})
+                    summary['updated' if existing else 'created'] += 1
+                    if post:
+                        move.action_post()
+                        summary['posted'] += 1
+                        self._fc_reconcile_payment(move, inv)
+            except Exception as e:  # noqa: BLE001 - per-invoice isolation
+                _logger.exception("Ledger ingest: invoice %s failed", nc_id)
+                summary['failed'].append({'id': nc_id, 'error': '%s: %s' % (type(e).__name__, e)})
+        return summary
+
+    @api.model
+    def _fc_reconcile_payment(self, move, inv):
+        """Placeholder until Task 5; defined so post=True doesn't AttributeError."""
+        return False
+```
+
+- [ ] **Step 4: run** → PASS. (If tax computes to 13.00 only when the company/fiscal position allows it, read the tax setup on trial; if `amount_tax` ≠ 13.00, the chosen tax is wrong — fix `_fc_tax_for`, never weaken the assertion.)
+
+- [ ] **Step 5: commit** — `feat(billing): ingest NexaCloud invoices -> draft account.move (idempotent)`
+
+---
+
+## Task 5: Reconcile Stripe payments (paid invoices show paid)
+
+**Read reference first:** confirm the payment-register flow on trial:
+```bash
+ssh pve-worker1 "qm guest exec 316 -- bash -lc 'docker exec odoo-trial-app bash -lc \"grep -nE \\\"_create_payments|def action_create_payments\\\" /mnt/enterprise-addons/account/wizard/account_payment_register.py | head\"'"
+```
+
+**Files:** modify `wizards/invoice_ledger.py`, `tests/test_invoice_ledger.py`.
+
+- [ ] **Step 1: failing test** (append):
+```python
+    def test_paid_invoice_is_reconciled_and_shows_paid(self):
+        data = _inv_fixture()
+        data[0].update({'status': 'paid', 'amount_paid': 113.0, 'paid_at': '2026-05-02'})
+        self.W._ingest_invoices(data, post=True)
+        mv = self.env['account.move'].search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
+        self.assertEqual(mv.state, 'posted')
+        self.assertIn(mv.payment_state, ('paid', 'in_payment'))
+```
+(Add this inside `TestLedgerIngest`.)
+
+- [ ] **Step 2: run** → FAIL (payment not reconciled).
+
+- [ ] **Step 3: implement** `_fc_reconcile_payment` + a journal helper (replace the placeholder):
+```python
+    @api.model
+    def _fc_stripe_journal(self):
+        Journal = self.env['account.journal']
+        j = Journal.search([('code', '=', 'NCSTR')], limit=1)
+        if not j:
+            j = Journal.create({'name': 'NexaCloud Stripe', 'code': 'NCSTR', 'type': 'bank'})
+        return j
+
+    @api.model
+    def _fc_reconcile_payment(self, move, inv):
+        paid = float(inv.get('amount_paid') or 0.0)
+        if (inv.get('status') != 'paid' and paid <= 0) or move.state != 'posted':
+            return False
+        reg = self.env['account.payment.register'].with_context(
+            active_model='account.move', active_ids=move.ids).create({
+                'journal_id': self._fc_stripe_journal().id,
+                'payment_date': inv.get('paid_at') or move.invoice_date or fields.Date.today(),
+                'amount': paid or move.amount_total,
+            })
+        reg._create_payments()
+        return True
+```
+
+- [ ] **Step 4: run** → PASS. (If `payment_state` is `in_payment` rather than `paid`, that's expected when the bank journal isn't reconciled to a statement — accept both, as the assertion does.)
+
+- [ ] **Step 5: commit** — `feat(billing): reconcile Stripe payments so ingested invoices show paid`
+
+---
+
+## Task 6: Reader + wizard actions + bulk-post + cron
+
+**Files:** modify `wizards/invoice_ledger.py`, `views/invoice_ledger_views.xml`, `tests/test_invoice_ledger.py`.
+
+- [ ] **Step 1: failing test** for bulk-post + DSN guard (append):
+```python
+    def test_post_ingested_posts_drafts(self):
+        self.W._ingest_invoices(_inv_fixture(), post=False)
+        n = self.W._post_ingested()
+        mv = self.env['account.move'].search([('x_fc_nexacloud_invoice_id', '=', 'inv-1')])
+        self.assertEqual(mv.state, 'posted')
+        self.assertGreaterEqual(n, 1)
+
+    def test_read_invoices_guards_missing_dsn(self):
+        from odoo.exceptions import UserError
+        self.env['ir.config_parameter'].sudo().set_param('fusion_billing.nexacloud_dsn', '')
+        with self.assertRaises(UserError):
+            self.W._read_nexacloud_invoices()
+```
+
+- [ ] **Step 2: run** → FAIL.
+
+- [ ] **Step 3: implement** `_post_ingested`, `_read_nexacloud_invoices`, `action_run`, and a cron entry:
+```python
+    @api.model
+    def _post_ingested(self):
+        moves = self.env['account.move'].search([
+            ('x_fc_nexacloud_invoice_id', '!=', False),
+            ('state', '=', 'draft'), ('move_type', '=', 'out_invoice')])
+        posted = 0
+        for mv in moves:
+            try:
+                with self.env.cr.savepoint():
+                    mv.action_post()
+                    posted += 1
+            except Exception as e:  # noqa: BLE001
+                _logger.exception("Ledger post: move %s failed", mv.id)
+        return posted
+
+    def _read_nexacloud_invoices(self, since=None):
+        import psycopg2
+        import psycopg2.extras
+        dsn = self.env['ir.config_parameter'].sudo().get_param('fusion_billing.nexacloud_dsn')
+        if not dsn:
+            raise UserError("NexaCloud DSN not configured (fusion_billing.nexacloud_dsn).")
+        try:
+            conn = psycopg2.connect(dsn)
+        except Exception as e:  # noqa: BLE001
+            raise UserError("Could not connect to the NexaCloud database: %s" % e)
+        try:
+            conn.set_session(readonly=True)
+            cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+            where = "WHERE i.created_at >= %(since)s" if since else ""
+            cur.execute(
+                "SELECT i.id, i.stripe_invoice_id, i.invoice_number, i.user_id AS user_external_id, "
+                "u.full_name AS partner_name, COALESCE(u.billing_email,u.email) AS partner_email, "
+                "i.created_at AS invoice_date, i.currency, i.status, i.subtotal, i.tax, "
+                "i.amount_paid, i.paid_at "
+                "FROM invoices i JOIN users u ON u.id = i.user_id " + where +
+                " ORDER BY i.created_at", {'since': since})
+            invoices = {str(r['id']): dict(r, items=[]) for r in cur.fetchall()}
+            cur.execute(
+                "SELECT ii.invoice_id, ii.description, ii.quantity, ii.unit_price, ii.amount "
+                "FROM invoice_items ii WHERE ii.invoice_id = ANY(%(ids)s)",
+                {'ids': list(invoices.keys())})
+            for r in cur.fetchall():
+                inv = invoices.get(str(r['invoice_id']))
+                if inv:
+                    inv['items'].append({'description': r['description'], 'quantity': r['quantity'],
+                                         'unit_price': r['unit_price'], 'amount': r['amount']})
+            for inv in invoices.values():
+                inv['id'] = str(inv['id'])
+                inv['user_external_id'] = str(inv['user_external_id'])
+            return list(invoices.values())
+        except psycopg2.Error as e:
+            raise UserError("Failed reading NexaCloud invoices — schema may have changed:\n%s" % e)
+        finally:
+            conn.close()
+
+    def action_run(self):
+        self.ensure_one()
+        data = self._read_nexacloud_invoices()
+        if self.dry_run:
+            class _Rollback(Exception):
+                pass
+            res = {}
+            try:
+                with self.env.cr.savepoint():
+                    res.update(self._ingest_invoices(data, post=False))
+                    raise _Rollback()
+            except _Rollback:
+                pass
+            res['dry_run'] = True
+        else:
+            res = self._ingest_invoices(data, post=self.auto_post)
+        self.result_summary = json.dumps(res, indent=2, default=str)
+        if res.get('failed'):
+            _logger.error("Ledger ingest: %s failed: %s", len(res['failed']), res['failed'])
+        return {"type": "ir.actions.act_window", "res_model": self._name,
+                "res_id": self.id, "view_mode": "form", "target": "new"}
+```
+Add a daily cron to `views/invoice_ledger_views.xml`:
+```xml
+    <record id="cron_fc_invoice_ledger" model="ir.cron">
+        <field name="name">Fusion Billing: Ingest NexaCloud invoices (daily)</field>
+        <field name="model_id" ref="model_fusion_billing_invoice_ledger_wizard"/>
+        <field name="state">code</field>
+        <field name="code">model.create({'dry_run': False, 'auto_post': True})._cron_ingest_recent()</field>
+        <field name="interval_number">1</field>
+        <field name="interval_type">days</field>
+        <field name="active">False</field>
+    </record>
+```
+And `_cron_ingest_recent` (ingest invoices from the last 2 days, idempotent):
+```python
+    def _cron_ingest_recent(self):
+        from datetime import timedelta
+        since = fields.Datetime.to_string(fields.Datetime.now() - timedelta(days=2))
+        return self._ingest_invoices(self._read_nexacloud_invoices(since=since), post=True)
+```
+(Cron ships `active=False` — enabled only after the backfill is reviewed.)
+
+- [ ] **Step 4: run** → PASS.
+
+- [ ] **Step 5: commit** — `feat(billing): invoice-ledger reader, wizard actions, bulk-post, daily cron`
+
+---
+
+## Task 7: Prune obsolete metered shadow data
+
+**Files:** modify `wizards/invoice_ledger.py`, `tests/test_invoice_ledger.py`.
+
+- [ ] **Step 1: failing test** (append):
+```python
+    def test_prune_shadow_removes_shadow_subs_only(self):
+        # a shadow sub + a normal order
+        p = self.env['res.partner'].sudo().create({'name': 'X'})
+        shadow = self.env['sale.order'].sudo().create({'partner_id': p.id, 'x_fc_shadow': True})
+        n = self.W._fc_prune_metered_shadow()
+        self.assertFalse(shadow.exists())
+        self.assertGreaterEqual(n.get('subscriptions', 0), 1)
+```
+
+- [ ] **Step 2: run** → FAIL.
+
+- [ ] **Step 3: implement**:
+```python
+    @api.model
+    def _fc_prune_metered_shadow(self):
+        """Delete the superseded metered shadow data (shadow sale.orders, NC-* products,
+        NexaCloud charges, reconciliation rows). Reversible only by re-import."""
+        counts = {}
+        subs = self.env['sale.order'].search([('x_fc_shadow', '=', True)])
+        counts['subscriptions'] = len(subs)
+        subs.unlink()
+        prods = self.env['product.product'].search([('default_code', '=like', 'NC-%')])
+        counts['products'] = len(prods)
+        prods.unlink()
+        ch = self.env['fusion.billing.charge'].search([])
+        counts['charges'] = len(ch)
+        ch.unlink()
+        rec = self.env['fusion.billing.reconciliation'].search([])
+        counts['reconciliations'] = len(rec)
+        rec.unlink()
+        return counts
+```
+
+- [ ] **Step 4: run** → PASS. (If a product can't unlink due to references, archive instead — read the error and adjust.)
+
+- [ ] **Step 5: commit** — `feat(billing): prune obsolete metered shadow data helper`
+
+---
+
+## Task 8: Full suite + static checks
+
+- [ ] `bash scripts/fcb_test_on_trial.sh` → `FCB_EXIT=0`.
+- [ ] `grep -rn "_sql_constraints" fusion_centralize_billing/ || echo clean` → clean.
+- [ ] `grep -rnE "sale\.subscription[^.]" fusion_centralize_billing/ | grep -v "sale.subscription.plan"` → only docstring.
+- [ ] commit any fixes.
+
+## Done = invoice ledger ready to run
+
+Then (separate, gated, NOT in this plan): on nexamain — prune shadow data, **dry-run** the full backfill (review the per-family $ summary + unmatched "Other" lines), ingest **as draft**, you review a sample, **bulk-post**, enable the daily cron.