Plano Base — Sistema de Disparo em Massa (ChatSkills Broadcast)

Estimativa total: 16 horas · Serviço independente que recebe uma requisição via API e dispara uma mensagem WhatsApp para todos os membros de um grupo, validando o owner da requisição e sua permissão sobre o grupo.

1. Contexto e Objetivo

O que existe hoje (`chatskills-api`)

O serviço atual (C:\Users\yuri_\projects\chatskills\chatskills-api) é um chatbot reativo: recebe webhooks do WhatsApp (Meta Cloud API) e responde a usuários que iniciaram a conversa.

Componentes reaproveitáveis como referência (não dependência direta):

Componente	Arquivo	O que reaproveitamos
Envio WhatsApp	`app/services/whatsapp/sender.py`	Padrão de POST para `graph.facebook.com/v19.0/{phone_number_id}/messages` com Bearer token
Auth/owner/grupo	`app/services/auth/user_auth.py`	Modelo de dados: `users`, `groups`, `group_members`, `teams`, `team_members(role=LEADER)`
Config	`app/core/config.py`	Pydantic Settings + `.env` (token Meta, DSN Postgres, Redis)
Migrations	`app/core/migrations.py` + `app/migrations/*.sql`	Migrations SQL idempotentes rodadas no startup
Infra	`docker-compose.yml`, `Dockerfile`, `captain-definition`	FastAPI + Postgres + Redis, deploy CapRover

O que vamos construir (`chatskills-broadcast`)

Serviço proativo e separado, com seu próprio app FastAPI, banco/tabelas próprias, fila e worker. Fluxo macro:

[Sistema externo: "novo conteúdo publicado p/ grupo X"]
        │  POST /api/v1/broadcasts  (API key + owner)
        ▼
  [API ingestão] ──► valida owner + permissão sobre o grupo
        │
        ▼
  [Resolve destinatários] ──► membros ACTIVE do grupo → telefones
        │
        ▼
  [Enfileira na Redis] ──► broadcast_recipients (status=pending)
        │
        ▼
  [Worker assíncrono] ──► envia Template Message (Meta API), com pacing/retry
        │
        ▼
  [Webhook de status] ──► sent / delivered / read / failed por destinatário

Restrição crítica de domínio (decisão de arquitetura)

Mensagens iniciadas pela empresa fora da janela de 24h de atendimento só podem ser enviadas como Template Messages (HSM) pré-aprovados pela Meta. Disparo de "nova atualização" é sempre business-initiated → o sistema envia templates aprovados, não texto livre. Implicações que o plano cobre:

Toda campanha referencia um template_name + language + variáveis.
Throughput limitado por tier de qualidade do número (1K/10K/100K/ilimitado por 24h) → pacing obrigatório.
Status de entrega vem por webhook assíncrono, não pela resposta do POST.

2. Arquitetura do Serviço

chatskills-broadcast/
├── app/
│   ├── main.py                      # FastAPI app + lifespan (DB/Redis/migrations)
│   ├── core/
│   │   ├── config.py                # Settings (token Meta, DSN, Redis, API keys)
│   │   ├── database.py              # asyncpg pool (PROD read + broadcast write)
│   │   ├── redis.py                 # cliente Redis (fila + rate-limit)
│   │   └── migrations.py            # runner idempotente
│   ├── migrations/
│   │   └── 001_broadcast.sql        # tabelas do serviço
│   ├── models/
│   │   └── schemas.py               # Pydantic: BroadcastIn, RecipientOut, etc.
│   ├── routers/
│   │   ├── broadcasts.py            # POST/GET /api/v1/broadcasts
│   │   └── status_webhook.py        # POST /webhooks/whatsapp-status
│   ├── services/
│   │   ├── auth/api_key.py          # autentica owner via API key
│   │   ├── permissions/resolver.py  # owner tem permissão sobre o grupo?
│   │   ├── audience/resolver.py     # grupo → telefones (membros ACTIVE)
│   │   ├── whatsapp/template_sender.py  # envio de Template Message + parse de erro
│   │   ├── queue/dispatcher.py      # enfileira + worker de consumo
│   │   └── broadcast/manager.py     # CRUD de campanhas e destinatários
├── tests/
├── docker-compose.yml
├── Dockerfile
├── captain-definition
├── requirements.txt
├── .env.example
└── docs/PLANO_DISPARO_MASSA.md

Decisões técnicas

Banco: tabelas próprias (broadcasts, broadcast_recipients, broadcast_api_keys) no mesmo Postgres do chatskills-api, isoladas por prefixo. Leitura read-only das tabelas de domínio (groups, group_members, team_members) para resolver audiência/permissão.
Fila: Redis (lista/stream) + worker asyncio no mesmo processo (KISS para v1). Evoluir para arq/worker dedicado se o volume exigir. Sem Celery na v1.
Idempotência: header Idempotency-Key por campanha + UNIQUE(broadcast_id, phone) em recipients.
Rate limiting: token-bucket em Redis (msgs/segundo configurável) respeitando o tier Meta.

3. Modelo de Dados (migration `001_broadcast.sql`)

-- Campanha de disparo
CREATE TABLE IF NOT EXISTS public.broadcasts (
    id              varchar(50) PRIMARY KEY,
    public_id       varchar(8)  NOT NULL UNIQUE,
    owner_user_id   varchar(50) NOT NULL,        -- quem disparou (resolvido da API key)
    group_id        varchar(50) NOT NULL,        -- grupo alvo
    template_name   varchar(120) NOT NULL,       -- HSM aprovado na Meta
    language        varchar(10)  NOT NULL DEFAULT 'pt_BR',
    variables       jsonb,                       -- parâmetros do template
    status          varchar(20)  NOT NULL DEFAULT 'queued', -- queued|sending|completed|failed|canceled
    total_count     int NOT NULL DEFAULT 0,
    sent_count      int NOT NULL DEFAULT 0,
    delivered_count int NOT NULL DEFAULT 0,
    failed_count    int NOT NULL DEFAULT 0,
    idempotency_key varchar(80),
    created_at      timestamp NOT NULL DEFAULT NOW(),
    updated_at      timestamp NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX IF NOT EXISTS broadcasts_idem_idx
    ON public.broadcasts(owner_user_id, idempotency_key)
    WHERE idempotency_key IS NOT NULL;

-- 1 linha por destinatário, com status individual
CREATE TABLE IF NOT EXISTS public.broadcast_recipients (
    id            varchar(50) PRIMARY KEY,
    broadcast_id  varchar(50) NOT NULL REFERENCES public.broadcasts(id) ON DELETE CASCADE,
    user_id       varchar(50),
    phone         varchar(20) NOT NULL,
    status        varchar(20) NOT NULL DEFAULT 'pending', -- pending|sent|delivered|read|failed
    wa_message_id varchar(80),                  -- id retornado pela Meta (correlaciona webhook)
    error_code    varchar(40),
    error_detail  text,
    attempts      int NOT NULL DEFAULT 0,
    updated_at    timestamp NOT NULL DEFAULT NOW(),
    UNIQUE (broadcast_id, phone)
);
CREATE INDEX IF NOT EXISTS broadcast_recipients_bid_idx ON public.broadcast_recipients(broadcast_id, status);
CREATE INDEX IF NOT EXISTS broadcast_recipients_wamid_idx ON public.broadcast_recipients(wa_message_id);

-- API keys + escopo de permissão do owner
CREATE TABLE IF NOT EXISTS public.broadcast_api_keys (
    id          varchar(50) PRIMARY KEY,
    owner_user_id varchar(50) NOT NULL,
    key_hash    varchar(80) NOT NULL UNIQUE,     -- hash da chave (nunca em texto)
    scopes      jsonb NOT NULL DEFAULT '["broadcast:send"]',
    active      boolean NOT NULL DEFAULT true,
    created_at  timestamp NOT NULL DEFAULT NOW()
);

Modelo de permissão: o owner pode disparar para um grupo se for LEADER ACTIVE do team dono do grupo (ou regra de ownership equivalente em groups/group_members). A query reaproveita a lógica de user_auth.py (team_members.role='LEADER' AND status='ACTIVE').

4. Contrato da API

`POST /api/v1/broadcasts`

Headers: Authorization: Bearer <api_key>, Idempotency-Key: <uuid> (opcional)

// Request
{
  "group_id": "grp_123",
  "template_name": "novo_conteudo_v1",
  "language": "pt_BR",
  "variables": { "1": "Curso de Python", "2": "https://app/curso/123" }
}
// Response 202 Accepted
{ "broadcast_id": "bc_...", "public_id": "8x4kT", "status": "queued", "total_recipients": 42 }

Validações: API key válida → resolve owner_user_id → permissão sobre group_id → resolve audiência → cria campanha + recipients (pending) → enfileira → responde 202.

`GET /api/v1/broadcasts/{id}` — progresso/contadores

`GET /api/v1/broadcasts/{id}/recipients?status=failed` — auditoria por destinatário

`POST /webhooks/whatsapp-status` — recebe status da Meta (sent/delivered/read/failed)

5. Cronograma de 16 Horas

#	Fase	Entregável	Horas
0	Scaffold & infra	Estrutura de pastas, `config.py`, `database.py`, `redis.py`, `main.py` com lifespan, `Dockerfile`, `docker-compose.yml`, `captain-definition`, `.env.example`, `requirements.txt`. App sobe em `/health`.	1.5
1	Modelo de dados	`001_broadcast.sql` + runner de migrations idempotente. Tabelas criadas no startup.	2.0
2	Auth do owner (API key)	`services/auth/api_key.py`: gera/valida key (hash), resolve `owner_user_id`. Dependency FastAPI que injeta o owner.	2.0
3	Permissão sobre o grupo	`services/permissions/resolver.py`: query read-only (LEADER ACTIVE / ownership). Nega 403 quando owner não tem direito ao grupo.	2.0
4	Resolver audiência	`services/audience/resolver.py`: grupo → telefones de membros ACTIVE, dedupe + validação de formato E.164.	1.5
5	API de ingestão	`routers/broadcasts.py`: `POST /api/v1/broadcasts` (idempotência, cria campanha+recipients, enfileira) + `GET` de status/recipients.	2.0
6	Worker de disparo	`services/queue/dispatcher.py` + `whatsapp/template_sender.py`: consome fila, envia Template Message, token-bucket (rate limit), retry/backoff, atualiza recipient.	2.5
7	Webhook de status	`routers/status_webhook.py`: correlaciona `wa_message_id` → atualiza status do recipient + contadores da campanha.	1.0
8	Testes & docs	Testes (permissão negada, idempotência, pacing, parse de webhook) + README/`.env.example`.	1.0
	TOTAL		16.0

Detalhamento por fase

Fase 0 — Scaffold & infra (1.5h)

Copiar padrões de chatskills-api: core/config.py (Settings com meta_whatsapp_token, meta_phone_number_id, db_prod_dsn, redis_url, broadcast_rate_per_sec).
main.py com lifespan inicializando Redis + pool Postgres + run_migrations().
Docker/CapRover espelhando o serviço atual (porta própria, ex. 8081).

Fase 1 — Modelo de dados (2h)

Escrever 001_broadcast.sql (seção 3). Garantir idempotência (IF NOT EXISTS).
broadcast/manager.py com create_broadcast, add_recipients, update_recipient_status, increment_counters (transações asyncpg como em support/manager.py).

Fase 2 — Auth do owner (2h)

api_key.py: hash_key(), verify_key(), resolve_owner(api_key) -> owner_user_id.
Dependency require_owner que retorna o owner ou 401. Script CLI simples p/ emitir keys.

Fase 3 — Permissão sobre o grupo (2h)

resolver.py: can_owner_broadcast(owner_user_id, group_id) -> bool reusando a regra de user_auth.py (LEADER ACTIVE). Cache curto em Redis (TTL ~60s) para evitar hit a cada request.

Fase 4 — Resolver audiência (1.5h)

resolver.py: resolve_phones(group_id) -> list[Recipient] (membros ACTIVE, telefone válido, dedupe). Tratar grupos grandes com fetch paginado.

Fase 5 — API de ingestão (2h)

Endpoint POST: orquestra fases 2→4, grava campanha+recipients, enfileira IDs na Redis, 202.
Idempotência via Idempotency-Key → se repetido, retorna a campanha existente.
GETs de status e de recipients (filtro por status).

Fase 6 — Worker de disparo (2.5h)

template_sender.py: payload type: "template" para a Meta API, parse de erros (ex. 131049 limite, 132xxx template), retorna wa_message_id.
dispatcher.py: loop assíncrono consumindo a fila, token-bucket em Redis para respeitar broadcast_rate_per_sec, retry com backoff (máx. N tentativas), marca sent/failed.

Fase 7 — Webhook de status (1h)

Recebe callbacks da Meta, extrai statuses[].id + status, faz match com wa_message_id, atualiza recipient e contadores agregados da campanha; conclui campanha quando todos resolvidos.

Fase 8 — Testes & docs (1h)

Testes: 403 sem permissão, idempotência, dedupe de audiência, rate-limit, parse de webhook (espelhar estilo de tests/test_support.py). README + .env.example.

6. Riscos & Mitigações

Risco	Impacto	Mitigação
Template não aprovado pela Meta	Disparo falha 100%	Validar `template_name` aprovado antes de aceitar a campanha; status de erro claro
Estouro de tier (rate limit Meta)	Mensagens rejeitadas (131049)	Token-bucket + pacing configurável; backoff em 131049
Telefones inválidos/duplicados	Falhas e custo desnecessário	Validação E.164 + dedupe na audiência
Reenvio acidental da mesma campanha	Spam ao grupo	`Idempotency-Key` + `UNIQUE(broadcast_id, phone)`
Acesso indevido a grupo de terceiros	Vazamento/abuso	Permissão obrigatória (LEADER ACTIVE) + API key com escopo
Webhook de status duplicado/fora de ordem	Contadores errados	Atualização idempotente por `wa_message_id` + transição de estado monotônica

7. Fora de Escopo (v1) / Próximos Passos

Agendamento de campanhas (cron) e janelas de envio.
Dashboard de campanhas (observabilidade segue como fase posterior).
Segmentação avançada (filtros por engajamento, tags).
Opt-out / gestão de consentimento (LGPD) — recomendado priorizar logo após a v1.
Multi-tenant de números WhatsApp (vários phone_number_id).

8. Pré-requisitos para começar

Token permanente Meta + phone_number_id (mesmo do chatskills-api ou dedicado).
Pelo menos um Template Message aprovado na Meta para os testes E2E.
Acesso read-only ao Postgres de PROD (groups, group_members, team_members).
Redis disponível (fila + rate-limit).
URL pública HTTPS para o webhook de status (CapRover ou ngrok em dev).

Plano base — ChatSkills Broadcast · revisão inicial · 2026-05-21