feat(billing): identity resolution external account -> partner

fusion_centralize_billing/models/account_link.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # Copyright 2026 Nexa Systems Inc.
 # License OPL-1
-from odoo import fields, models
+from odoo import api, fields, models
 
 
 class FusionBillingAccountLink(models.Model):
@@ -31,3 +31,27 @@ class FusionBillingAccountLink(models.Model):
         "unique(service_id, external_id)",
         "An external account can only link to one partner per service.",
     )
+
+    @api.model
+    def _resolve_or_create_partner(self, service, external_id, name=None, email=None, extra=None):
+        """Return the link for (service, external_id), creating partner+link if needed.
+
+        Unifies customers: if a link for this external_id exists, reuse it; else if a
+        partner with the same email already exists (possibly from another service),
+        link to it; else create a new partner.
+        """
+        existing = self.search(
+            [('service_id', '=', service.id), ('external_id', '=', external_id)], limit=1)
+        if existing:
+            return existing
+        partner = self.env['res.partner']
+        if email:
+            partner = partner.search([('email', '=', email)], limit=1)
+        if not partner:
+            partner = partner.create({'name': name or external_id, 'email': email, **(extra or {})})
+        return self.create({
+            'service_id': service.id,
+            'external_id': external_id,
+            'external_email': email,
+            'partner_id': partner.id,
+        })

fusion_centralize_billing/tests/test_identity.py

@@ -23,3 +23,33 @@ class TestServiceApiKey(TransactionCase):
         self.assertFalse(self.Service._match_api_key('nope-not-a-key'))
         self.service.active = False
         self.assertFalse(self.Service._match_api_key(raw))
+
+
+@tagged('post_install', '-at_install')
+class TestIdentityResolution(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.service = self.env['fusion.billing.service'].sudo().create(
+            {'name': 'NexaDesk', 'code': 'nexadesk'})
+        self.Link = self.env['fusion.billing.account.link'].sudo()
+
+    def test_creates_partner_first_time(self):
+        link = self.Link._resolve_or_create_partner(
+            self.service, external_id='tenant-1', name='Acme Inc', email='ar@acme.test')
+        self.assertTrue(link.partner_id)
+        self.assertEqual(link.partner_id.name, 'Acme Inc')
+        self.assertEqual(link.external_id, 'tenant-1')
+
+    def test_idempotent_same_external_id(self):
+        a = self.Link._resolve_or_create_partner(self.service, 'tenant-1', 'Acme', 'ar@acme.test')
+        b = self.Link._resolve_or_create_partner(self.service, 'tenant-1', 'Acme Renamed', 'ar@acme.test')
+        self.assertEqual(a, b)                      # same link row
+        self.assertEqual(a.partner_id, b.partner_id)  # same partner
+
+    def test_reuses_partner_by_email_across_services(self):
+        other = self.env['fusion.billing.service'].sudo().create({'name': 'Maps', 'code': 'nexamaps'})
+        a = self.Link._resolve_or_create_partner(self.service, 'tenant-1', 'Acme', 'ar@acme.test')
+        b = self.Link._resolve_or_create_partner(other, 'client-9', 'Acme', 'ar@acme.test')
+        self.assertEqual(a.partner_id, b.partner_id)  # one unified customer
+        self.assertNotEqual(a, b)                     # but distinct link rows