feat(comunicados): pipeline de sincronización de comunicados hacia c_comunicados

Un documento por comunicado con su regla de audiencia (alcance_tipo y
arreglos sede_ids, aula_ids, matricula_ids) en vez de una fila por alumno.
Incluye FetchByID para el upsert puntual que dispara Trismegisto.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -51,6 +51,7 @@ All endpoints prefixed with `/api/v1`.
 - `POST /api/v1/sync/parents/trigger` / `GET /api/v1/sync/parents/status` — parents pipeline (same shape).
 - `POST /api/v1/sync/users/trigger` / `GET /api/v1/sync/users/status` — users pipeline (same shape).
 - `POST /api/v1/sync/balotarios/trigger` / `GET /api/v1/sync/balotarios/status` — pipeline de balotarios (misma forma). La primera corrida tarda ~2 min, así que usar `curl -m 290`.
+- `POST /api/v1/sync/comunicados/trigger` / `GET /api/v1/sync/comunicados/status` — pipeline de comunicados (misma forma). Corrida completa barata: un documento por comunicado, no por alumno.
 - `POST /api/v1/sync/users/upsert` — upsert puntual de un solo usuario, sin esperar al ciclo programado. Body `{"id": <user_id>}`; relee esa fila de `intranet.users` en Postgres y la refleja en la colección `users` de Mongo. Responde `{"status": true, "module": "users", "id": N, "outcome": "created|updated|unchanged"}`, `404` si el id no existe en el origen. Lo consume Trismegisto al crear usuario o restablecer contraseña desde `padresIntranet.jsp`. Solo se registra para los módulos con lector por id en `pipeline.rowReaderFor` (hoy `users`); agregar un `case` ahí más un `FetchByID` en el reader habilita el endpoint para otro módulo.
 
 ## Postgres source contract
@@ -65,6 +66,8 @@ Users: `internal/infrastructure/database/postgres.UserQuery`, one row per `matri
 
 Balotarios: `internal/infrastructure/database/postgres.BalotarioQuery`, un documento por matrícula activa del período 14 / ciclo 28 (~35 700), con los 8 períodos pedagógicos (M1, B1, M2, B2, M3, B3, M4, B4) siempre presentes y sus cursos anidados (`cursos: []` si no hay nada configurado). Los cursos salen de `academico.ac_balotario_aula` por el aula actual **más** los balotarios que el alumno ya respondió (por si cambió de aula); **no** se filtra por `grado_id`, porque las aulas multigrado y "PRE" reciben a propósito los balotarios de otro grado. Leído por `postgres.BalotarioReader` hacia `entities.Balotario`: colección `c_balotario`, archivo `deleted_c_balotario`, `IDField` = `matricula_id`, `StateID` = 6. Lo consume `GET /enrollments/:id/balotarios` de intranet-so-be. Detalle y hallazgos de datos en `intranet-padres-react/docs/backend-queries.md`.
 
+Comunicados: `internal/infrastructure/database/postgres.ComunicadoQuery`, un documento por comunicado (`intranet.comunicado` + `intranet.comunicado_destino`) con su regla de audiencia, no una fila por alumno: `alcance_tipo` (1 todos, 2 sedes, 3 aulas, 4 personalizado) y los arreglos `sede_ids`, `aula_ids` y `matricula_ids`. Se sincronizan también los comunicados vencidos o retirados, porque intranet-so-be filtra por `vigente_hasta` al leer y así retirar uno surte efecto al instante. Leído por `postgres.ComunicadoReader` hacia `entities.Comunicado`: colección `c_comunicados`, archivo `deleted_c_comunicados`, `IDField` = `comunicado_id`, `StateID` = 11 (el 10 lo usa la rama `pipelines-libreta`). Requiere índice único sobre `comunicado_id` en Mongo. Trismegisto dispara `POST /api/v1/sync/comunicados/upsert` con `{"id": <comunicado_id>}` al enviar o retirar. El DDL y las funciones están en `intranet-so-be/migrations/004_comunicados.sql`.
+
 ## Mongo `_id` rule
 
 **`_id` nunca es el id de negocio.** Todas las colecciones dejan que Mongo genere su propio ObjectID; el id de negocio (`student_id`, `payment_plan_id`, etc.) vive como campo normal del documento — el mismo declarado en `PipelineDef.IDField` (`internal/infrastructure/config/pipelines.go`). `repository.DocumentRepository` y `repository.DeletedRepository` reciben ese `idField` y filtran/listan/archivan por él, no por `_id`. Al archivar en la colección `deleted_*`, se descarta el `_id` original para que se genere uno nuevo. Cualquier pipeline nuevo debe seguir esta misma regla — no volver a usar `_id = <id de negocio>`. Requiere índice único sobre `idField` en cada colección Mongo (no lo impone el código).

internal/domain/entities/comunicado.go

+package entities
+
+import "time"
+
+type Comunicado struct {
+	ComunicadoID    int       `bson:"comunicado_id"`
+	Tipo            string    `bson:"tipo"`
+	Titulo          string    `bson:"titulo"`
+	Cuerpo          string    `bson:"cuerpo"`
+	EvidenciaURL    string    `bson:"evidencia_url"`
+	EvidenciaNombre string    `bson:"evidencia_nombre"`
+	AlcanceTipo     int       `bson:"alcance_tipo"`
+	Fecha           string    `bson:"fecha"`
+	VigenteHasta    string    `bson:"vigente_hasta"`
+	RemitenteNombre string    `bson:"remitente_nombre"`
+	SedeIDs         []int32   `bson:"sede_ids"`
+	AulaIDs         []int32   `bson:"aula_ids"`
+	MatriculaIDs    []int32   `bson:"matricula_ids"`
+	CreatedAt       time.Time `bson:"created_at"`
+	UpdatedAt       time.Time `bson:"updated_at"`
+	RowHash         string    `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (c *Comunicado) SetUpdatedAt(t time.Time) {
+	c.UpdatedAt = t
+}
+
+func (c *Comunicado) SetRowHash(h string) {
+	c.RowHash = h
+}
+
+func (c *Comunicado) SetCreatedAt(t time.Time) {
+	c.CreatedAt = t
+}

internal/infrastructure/config/pipelines.go

@@ -84,4 +84,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "matricula_id",
 		StateID:                9,
 	},
+	{
+		Name:                   "comunicados",
+		PGQuery:                postgres.ComunicadoQuery,
+		MongoCollection:        "c_comunicados",
+		MongoDeletedCollection: "deleted_c_comunicados",
+		IDField:                "comunicado_id",
+		StateID:                11,
+	},
 }

internal/infrastructure/database/postgres/comunicado_query.go

+package postgres
+
+// ComunicadoQuery arma un documento por comunicado con su regla de audiencia, no uno por alumno:
+// alcance_tipo (1 todos, 2 sedes, 3 aulas, 4 personalizado) y los arreglos de destinos sede_ids,
+// aula_ids y matricula_ids. Se incluyen también los comunicados vencidos o retirados, porque
+// intranet-so-be filtra por vigente_hasta al leer y así retirar un comunicado surte efecto en cuanto
+// Trismegisto dispara el upsert puntual.
+const comunicadoSelect = `
+SELECT c.comunicado_id,
+       c.comunicado_tipo,
+       c.comunicado_titulo,
+       c.comunicado_cuerpo,
+       c.comunicado_evidencia_url,
+       c.comunicado_evidencia_nombre,
+       c.alcance_tipo,
+       to_char(c.comunicado_fecha_registro, 'YYYY-MM-DD"T"HH24:MI:SS') AS fecha,
+       to_char(c.comunicado_vigente_hasta, 'YYYY-MM-DD') AS vigente_hasta,
+       (SELECT UPPER(COALESCE(pp.persona_nombre, '') || ' ' || COALESCE(pp.persona_apellido_paterno, '') || ' ' || COALESCE(pp.persona_apellido_materno, ''))
+          FROM administracion.ad_usuario au
+                   JOIN administracion.ad_empleado ae ON ae.empleado_id = au.empleado_id
+                   JOIN persona.pe_persona pp ON pp.persona_id = ae.persona_id
+         WHERE au.usuario_id = c.usuario_id_remitente) AS remitente_nombre,
+       COALESCE(ARRAY_AGG(d.destino_id ORDER BY d.destino_id) FILTER (WHERE d.destino_tipo = 'S'), '{}'::int[]) AS sede_ids,
+       COALESCE(ARRAY_AGG(d.destino_id ORDER BY d.destino_id) FILTER (WHERE d.destino_tipo = 'A'), '{}'::int[]) AS aula_ids,
+       COALESCE(ARRAY_AGG(d.destino_id ORDER BY d.destino_id) FILTER (WHERE d.destino_tipo = 'M'), '{}'::int[]) AS matricula_ids
+FROM intranet.comunicado c
+         LEFT JOIN intranet.comunicado_destino d ON d.comunicado_id = c.comunicado_id
+`
+
+const comunicadoGroupBy = `
+GROUP BY c.comunicado_id
+`
+
+const ComunicadoQuery = comunicadoSelect + comunicadoGroupBy + `ORDER BY c.comunicado_id
+`
+
+// ComunicadoByIDQuery es ComunicadoQuery acotado a un solo comunicado, para la carga parcial
+// que Trismegisto dispara al enviar o retirar uno (POST /api/v1/sync/comunicados/upsert).
+const ComunicadoByIDQuery = comunicadoSelect + `WHERE c.comunicado_id = $1
+` + comunicadoGroupBy

internal/infrastructure/database/postgres/comunicado_reader.go

+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+)
+
+type ComunicadoReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewComunicadoReader(pool *Pool, query string) repositories.SourceReader {
+	return &ComunicadoReader{pool: pool, query: query}
+}
+
+func NewComunicadoRowReader(pool *Pool) repositories.SourceRowReader {
+	return &ComunicadoReader{pool: pool, query: ComunicadoByIDQuery}
+}
+
+func scanComunicadoRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var (
+		comunicadoID    int
+		tipo            string
+		titulo          string
+		cuerpo          string
+		evidenciaURL    *string
+		evidenciaNombre *string
+		alcanceTipo     int
+		fecha           string
+		vigenteHasta    *string
+		remitenteNombre *string
+		sedeIDs         []int32
+		aulaIDs         []int32
+		matriculaIDs    []int32
+	)
+	if err := row.Scan(&comunicadoID, &tipo, &titulo, &cuerpo, &evidenciaURL, &evidenciaNombre,
+		&alcanceTipo, &fecha, &vigenteHasta, &remitenteNombre, &sedeIDs, &aulaIDs, &matriculaIDs); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan comunicado row: %w", err)
+	}
+
+	doc := &entities.Comunicado{
+		ComunicadoID:    comunicadoID,
+		Tipo:            tipo,
+		Titulo:          titulo,
+		Cuerpo:          cuerpo,
+		EvidenciaURL:    derefString(evidenciaURL),
+		EvidenciaNombre: derefString(evidenciaNombre),
+		AlcanceTipo:     alcanceTipo,
+		Fecha:           fecha,
+		VigenteHasta:    derefString(vigenteHasta),
+		RemitenteNombre: derefString(remitenteNombre),
+		SedeIDs:         nonNilIDs(sedeIDs),
+		AulaIDs:         nonNilIDs(aulaIDs),
+		MatriculaIDs:    nonNilIDs(matriculaIDs),
+	}
+	return entities.SyncRow{ID: comunicadoID, Doc: doc}, nil
+}
+
+func nonNilIDs(ids []int32) []int32 {
+	if ids == nil {
+		return []int32{}
+	}
+	return ids
+}
+
+func (f *ComunicadoReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query comunicados: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanComunicadoRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate comunicado rows: %w", err)
+	}
+	return result, nil
+}
+
+func (f *ComunicadoReader) FetchByID(ctx context.Context, id int) (entities.SyncRow, bool, error) {
+	rows, err := f.pool.Pool.Query(ctx, ComunicadoByIDQuery, id)
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("query comunicado %d: %w", id, err)
+	}
+
+	row, err := pgx.CollectExactlyOneRow(rows, scanComunicadoRow)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return entities.SyncRow{}, false, nil
+	}
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("read comunicado %d: %w", id, err)
+	}
+	return row, true, nil
+}

internal/infrastructure/pipeline/builder.go

@@ -34,6 +34,8 @@ func readerFor(name string, pool *postgres.Pool, query string) (repositories.Sou
 		return postgres.NewHorarioReader(pool, query), nil
 	case "mensajes":
 		return postgres.NewMensajeReader(pool, query), nil
+	case "comunicados":
+		return postgres.NewComunicadoReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}
@@ -51,6 +53,8 @@ func rowReaderFor(name string, pool *postgres.Pool) repositories.SourceRowReader
 		return postgres.NewHorarioRowReader(pool)
 	case "mensajes":
 		return postgres.NewMensajeRowReader(pool)
+	case "comunicados":
+		return postgres.NewComunicadoRowReader(pool)
 	default:
 		return nil
 	}