Odoo-Modules/fusion_plating/fusion_plating/models/fp_parent_numbered_mixin.py

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""Abstract mixin: derive a record's name from its parent sale order.

Every model that 1:1 links to an SO inherits this mixin. The mixin
owns the atomic counter logic so race conditions and counter drift
are impossible. Subclasses implement three small hooks and call
``self._fp_assign_parent_name()`` from their ``create()`` override.

See docs/superpowers/specs/2026-05-12-parent-number-hierarchy-design.md
for the design rationale.
"""
import re

from markupsafe import Markup
from odoo import fields, models
from odoo.exceptions import UserError
from odoo.tools.translate import _

# Whitelist regex for counter-field names. The mixin interpolates the
# returned name into raw SQL, so a future subclass that read this from
# a context value or Selection field would otherwise open a SQL-injection
# surface. Enforce: must look like one of our x_fc_pn_*_count counters
# (lowercase letters / underscores only).
_FP_COUNTER_FIELD_RE = re.compile(r'^x_fc_pn_[a-z_]+_count$')


class FpParentNumberedMixin(models.AbstractModel):
    _name = 'fp.parent.numbered.mixin'
    _description = 'Fusion Plating - Parent-Number-Derived Naming'

    x_fc_doc_index = fields.Integer(
        string='Parent Doc Index',
        readonly=True,
        copy=False,
        index=True,
        help='1-based position within this parent SO. 1 = the first '
             'child of this type for the SO; subsequent siblings get 2, '
             '3, etc. The first sibling renders its name bare; later '
             'siblings get a zero-padded "-NN" suffix.',
    )

    # ------------------------------------------------------------------
    # Hooks subclasses must override
    # ------------------------------------------------------------------
    def _fp_parent_sale_order(self):
        """Return the linked sale.order recordset (or empty)."""
        return self.env['sale.order']

    def _fp_name_prefix(self):
        """Return the model's prefix (e.g. 'WO', 'IN', 'CoC')."""
        raise NotImplementedError(
            'Subclass must define _fp_name_prefix()'
        )

    def _fp_parent_counter_field(self):
        """Return the counter field on sale.order for THIS model."""
        raise NotImplementedError(
            'Subclass must define _fp_parent_counter_field()'
        )

    # ------------------------------------------------------------------
    # Core: atomic counter + name composition
    # ------------------------------------------------------------------
    def _fp_compose_name(self, parent_number, index):
        """Pure helper: compose the name string per the design's rules."""
        prefix = self._fp_name_prefix()
        if index <= 1:
            return f'{prefix}-{parent_number}'
        if index <= 99:
            return f'{prefix}-{parent_number}-{index:02d}'
        return f'{prefix}-{parent_number}-{index}'

    def _fp_assign_parent_name(self):
        """Lock the parent SO, bump the counter, set name + doc index.

        Returns True if assignment succeeded; False if no parent SO is
        linked (caller falls back to the model's own legacy sequence).
        """
        self.ensure_one()
        so = self._fp_parent_sale_order()
        # Defensive: the parent-number column lives in fusion_plating_jobs;
        # downstream modules (e.g. fusion_plating_receiving) inherit the
        # mixin but don't depend on jobs, so so.x_fc_parent_number can
        # raise AttributeError at test time. hasattr keeps the mixin safe
        # in either install topology — falls through to the legacy
        # sequence when the column isn't there.
        if not so or 'x_fc_parent_number' not in so._fields:
            return False
        if not so.x_fc_parent_number:
            return False
        counter_field = self._fp_parent_counter_field()
        # Whitelist check — the field name is interpolated directly into
        # SQL below, so we never trust an arbitrary string. All current
        # subclasses return a literal; this guard exists so a future
        # subclass that reads the field name from context / Selection /
        # user input can't smuggle a SQL fragment in.
        if not _FP_COUNTER_FIELD_RE.match(counter_field or ''):
            raise UserError(_(
                'Invalid parent-counter field name %r — must match '
                'pattern x_fc_pn_*_count.'
            ) % counter_field)
        # SELECT FOR UPDATE - locks the SO row until commit, so a
        # concurrent create on the same SO blocks here and reads the
        # updated counter after we release. No race, no drift.
        self.env.cr.execute(
            f'SELECT {counter_field} FROM sale_order WHERE id = %s FOR UPDATE',
            (so.id,),
        )
        row = self.env.cr.fetchone()
        current = (row and row[0]) or 0
        new_index = current + 1
        self.env.cr.execute(
            f'UPDATE sale_order SET {counter_field} = %s WHERE id = %s',
            (new_index, so.id),
        )
        so.invalidate_recordset([counter_field])
        new_name = self._fp_compose_name(so.x_fc_parent_number, new_index)
        # Raw SQL update bypasses the immutability write() guard added
        # in Task 11 (since this IS the legitimate assignment path).
        self.env.cr.execute(
            f'UPDATE {self._table} SET name = %s, x_fc_doc_index = %s WHERE id = %s',
            (new_name, new_index, self.id),
        )
        self.invalidate_recordset(['name', 'x_fc_doc_index'])
        so.message_post(body=Markup(_(
            'Issued <strong>%s</strong> to %s #%s.'
        )) % (new_name, self._name, self.id))
        return True

    # ------------------------------------------------------------------
    # Immutability: name + x_fc_doc_index can't change post-issuance.
    # Bypass: context flag fp_allow_name_rename=True. Used ONLY by:
    #   1. sale.order.action_confirm (Q -> SO rename, one-time)
    #   2. Bulk WO creation mid-create (sets names explicitly)
    #   3. Legacy-sequence fallback path in child create() overrides
    # Compliance: once issued, an audit-trail number can never change.
    # ------------------------------------------------------------------
    FP_IMMUTABLE_FIELDS = ('name', 'x_fc_doc_index')

    def write(self, vals):
        if not self.env.context.get('fp_allow_name_rename'):
            for f in self.FP_IMMUTABLE_FIELDS:
                if f in vals:
                    for rec in self:
                        current = rec[f]
                        if current and current != vals[f]:
                            raise UserError(_(
                                'Field "%(field)s" on %(model)s "%(name)s" '
                                'is immutable. Once issued, it cannot be '
                                'changed - this preserves the compliance '
                                'audit trail. (Attempted: %(old)r -> %(new)r)'
                            ) % {
                                'field': f, 'model': self._description,
                                'name': rec.display_name,
                                'old': current, 'new': vals[f],
                            })
        return super().write(vals)

    # ------------------------------------------------------------------
    # Unlink block: issued documents can't be hard-deleted.
    # Cancellation must go through the state machine so the audit trail
    # keeps the issued number tied to its cancellation reason. Hard
    # delete would leave a phantom gap in the counter. Applies to ALL
    # users including admins — no group bypass.
    # ------------------------------------------------------------------
    def unlink(self):
        for rec in self:
            # Records still in their initial 'New' state (no number
            # ever issued) are fine to delete — they're not yet in
            # the audit trail. Once x_fc_doc_index is non-zero OR
            # name is something other than 'New' / '/', the record
            # has been issued and is permanent.
            issued = rec.x_fc_doc_index or (
                rec.name and rec.name not in (False, '', 'New', '/')
            )
            if issued:
                raise UserError(_(
                    'Document "%(name)s" cannot be deleted - it is '
                    'part of the compliance audit trail. Cancel it '
                    'instead (use the state machine\'s Cancel action). '
                    'This rule applies to all users including '
                    'administrators.'
                ) % {'name': rec.display_name})
        return super().unlink()