async def materialize_campaign(db: AsyncSession, campaign: Campaign, recipient_rows: Sequence[Recipient],
                               preview: bool = False, preview_size: int = 0) -> dict:
    """Create CampaignRecipient rows with generated values & assignments.

    This is the core deterministic plan: which subject/template/phone/account each recipient gets,
    plus generated invoice numbers (unique per campaign), amounts, shipping.
    """
    subjects, templates, phones, accounts = await get_campaign_entities(db, campaign)

    var_cfg = dict(campaign.variable_config or {})
    engine = VariableEngine(var_cfg)

    subj_sel = SubjectSelector(subjects, campaign.subject_rotation)
    tpl_seq = TemplateSequencer(templates, campaign.template_sequence_mode)
    phone_sel = PhoneSelector(phones, campaign.phone_selection)
    rot = RotationAssigner(accounts, campaign.rotation_strategy, campaign.rotation_weights,
                           campaign.max_per_account)

    used_invoice_numbers: set = set()
    rows: List[CampaignRecipient] = []
    plan = []
    seq = 0
    for r in recipient_rows:
        seq += 1
        cr = CampaignRecipient(
            campaign_id=campaign.id,
            recipient_id=r.id,
            email=r.email,
            seq=seq,
        )
        # generated values
        g = GeneratedValues()
        if var_cfg.get("invoice_date"):
            g.invoice_date = var_cfg["invoice_date"]
        if var_cfg.get("due_date"):
            g.due_date = var_cfg["due_date"]
        inv_num = ""
        tpl = tpl_seq.next_template()
        if tpl is None and templates:
            cr.status = "skipped"
            cr.last_error = "Template sequence exhausted (mode=stop)"
        body_uses_invoice = any("%RandomInvoiceNumber%" in (t.body_html + (t.body_text or "")) for t in templates) or (tpl and "%RandomInvoiceNumber%" in tpl.body_html)
        if body_uses_invoice or var_cfg.get("always_generate_invoice", True):
            inv_num = engine.generate_invoice_number(used_invoice_numbers)
            used_invoice_numbers.add(inv_num)
        cr.generated_invoice_number = inv_num
        g.invoice_number = inv_num

        if var_cfg.get("amount_enabled", True) or body_uses_invoice:
            g.amount = engine.generate_amount()
            cr.generated_amount = g.amount
        if var_cfg.get("shipping_enabled", True) or "%RandomShippingCharge%" in (tpl.body_html if tpl else ""):
            g.shipping = engine.generate_shipping()
            cr.generated_shipping = g.shipping

        subj = subj_sel.next_subject()
        cr.generated_subject_id = subj.id if subj else None
        cr.generated_template_id = tpl.id if tpl else None

        phone = phone_sel.next_phone()
        cr.generated_phone = phone
        g.phone = phone

        account = rot.next_account()
        if account is None:
            cr.status = "skipped"
            cr.last_error = "No available sender account (limits or cooldown)"
        else:
            cr.assigned_account_id = account.id
        rows.append(cr)
        plan.append({"seq": seq, "email": r.email, "invoice": inv_num, "amount": g.amount,
                     "subject": subj.subject if subj else "", "template": tpl.name if tpl else None,
                     "phone": phone, "account": account.email if account else None,
                     "account_id": account.id if account else None,
                     "recipient_id": r.id, "subject_id": subj.id if subj else None,
                     "template_id": tpl.id if tpl else None,
                     "invoice_number": inv_num,
                     "invoice_date": g.invoice_date, "due_date": g.due_date,
                     "shipping": g.shipping,
                     "warnings": cr.last_error if cr.status == "skipped" else "",
                     "status": cr.status})
        if preview and seq >= (preview_size or campaign.preview_sample_size or 5):
            break

    if not preview:
        db.add_all(rows)
        await db.flush()
    return {"rows": rows, "plan": plan, "rotation": rot.status(),
            "account_count": len(accounts), "subjects": len(subjects),
            "templates": len(templates), "phones": len(phones)}