diff --git a/docs/superpowers/plans/2026-05-27-nexacloud-invoice-ledger.md b/docs/superpowers/plans/2026-05-27-nexacloud-invoice-ledger.md new file mode 100644 index 00000000..e6225a7b --- /dev/null +++ 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 + + + + fusion.billing.invoice.ledger.wizard.form + fusion.billing.invoice.ledger.wizard + +
+ + + + + + + +
+
+
+ + + + Ingest NexaCloud Invoices + fusion.billing.invoice.ledger.wizard + form + new + + + +``` + +- [ ] **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 + + Fusion Billing: Ingest NexaCloud invoices (daily) + + code + model.create({'dry_run': False, 'auto_post': True})._cron_ingest_recent() + 1 + days + False + +``` +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.