merge: integrar pipeline de balotarios

CLAUDE.md

@@ -50,6 +50,7 @@ All endpoints prefixed with `/api/v1`.
 - `POST /api/v1/sync/payment_plans/trigger` / `GET /api/v1/sync/payment_plans/status` — payment_plans pipeline (same shape).
 - `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/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
@@ -62,6 +63,8 @@ Parents: `internal/infrastructure/database/postgres.ParentQuery`, joins `persona
 
 Users: `internal/infrastructure/database/postgres.UserQuery`, one row per `matricula.ma_usuario`-equivalent login (`user_id`, `user_login`, `user_password`, `user_creation_date`, `parent_id`, `user_status`). Read by `postgres.UserReader` into `entities.User` — collection `users`, archive `deleted_users`, `IDField` = `user_id`, `StateID` = 4.
 
+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`.
+
 ## 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).
@@ -94,6 +97,7 @@ El `row_hash` se calcula sobre el JSON del documento, así que **cualquier array
 Implementado hoy en:
 - `parent_query.go` — `JSONB_AGG(DISTINCT jsonb_build_object('student_id', ...) ORDER BY ...)`, más `sort.Slice` por `StudentID` en `parent_reader.go`.
 - `payment_plan_query.go` — `ORDER BY payment_plan_order, payment_plan_id` en el `SELECT` externo (orden funcional por número de orden, con el id como desempate determinista); los años de `payment_plans` se ordenan descendente en `buildPaymentPlanYears`.
+- `balotario_query.go` — períodos con `ORDER BY p.periodo_orden` y cursos con `ORDER BY curso_id` dentro de cada `JSONB_AGG`, más `sortBalotarioPeriodos` en `balotario_reader.go` (períodos por `periodo_orden`, cursos por `curso_id`). El orden alfabético de cursos lo resuelve el frontend.
 
 Cualquier módulo nuevo con arrays anidados debe cumplir esta regla antes de entrar a `config.Pipelines`.
 
@@ -104,7 +108,7 @@ Misma arquitectura por capas que `intranet-so-be`: `domain` (entidades + puertos
 ```
 cmd/server/main.go                  raíz de composición (config -> DBs -> pipelines -> router -> HTTP)
 pkg/jsonutil/                       helpers genéricos (DecodeList para columnas JSON_AGG)
-internal/domain/entities/           Student, Parent, User, Professor, SyncRow, SyncResult
+internal/domain/entities/           Student, Parent, User, Professor, Balotario, SyncRow, SyncResult
 internal/domain/repositories/       puertos: SourceReader, DocumentUpserter, DocumentLister,
                                     DeletedArchiver, SyncStateRepository
 internal/usecase/                   RunSyncPipelineUseCase (ciclo completo), Scheduler, Schedule

internal/domain/entities/balotario.go

+package entities
+
+import "time"
+
+type BalotarioCurso struct {
+	BalotarioID   int     `bson:"balotario_id" json:"balotario_id"`
+	CursoID       int     `bson:"curso_id" json:"curso_id"`
+	CursoNombre   *string `bson:"curso_nombre" json:"curso_nombre"`
+	Responde      bool    `bson:"responde" json:"responde"`
+	Puntuacion    *string `bson:"puntuacion" json:"puntuacion"`
+	FormularioURL *string `bson:"formulario_url" json:"formulario_url"`
+}
+
+type BalotarioPeriodo struct {
+	PeriodoID     int              `bson:"periodo_id" json:"periodo_id"`
+	PeriodoNombre string           `bson:"periodo_nombre" json:"periodo_nombre"`
+	PeriodoOrden  int              `bson:"periodo_orden" json:"periodo_orden"`
+	Cursos        []BalotarioCurso `bson:"cursos" json:"cursos"`
+}
+
+type Balotario struct {
+	MatriculaID        int                `bson:"matricula_id"`
+	EstudianteID       int                `bson:"estudiante_id"`
+	PeriodoAcademicoID int                `bson:"periodo_academico_id"`
+	Periodos           []BalotarioPeriodo `bson:"periodos"`
+	CreatedAt          time.Time          `bson:"created_at"`
+	UpdatedAt          time.Time          `bson:"updated_at"`
+	RowHash            string             `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (r *Balotario) SetUpdatedAt(t time.Time) {
+	r.UpdatedAt = t
+}
+
+func (r *Balotario) SetRowHash(h string) {
+	r.RowHash = h
+}
+
+func (r *Balotario) SetCreatedAt(t time.Time) {
+	r.CreatedAt = t
+}

internal/infrastructure/config/pipelines.go

@@ -52,4 +52,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "professor_id",
 		StateID:                5,
 	},
+	{
+		Name:                   "balotarios",
+		PGQuery:                postgres.BalotarioQuery,
+		MongoCollection:        "c_balotario",
+		MongoDeletedCollection: "deleted_c_balotario",
+		IDField:                "matricula_id",
+		StateID:                6,
+	},
 }

internal/infrastructure/database/postgres/balotario_query.go

+package postgres
+
+const BalotarioQuery = `
+WITH periodos (periodo_id, periodo_orden) AS (
+    VALUES (3170, 1), (3174, 2), (3171, 3), (3175, 4), (3172, 5), (3176, 6), (3173, 7), (3177, 8)
+), alumnos AS (
+    SELECT m.matricula_id, m.estudiante_id, m.aula_id_asiste AS aula_id
+    FROM matricula.ma_matricula m
+             INNER JOIN academico.ac_aula au ON au.aula_id = m.aula_id_asiste
+    WHERE NOT m.matricula_retirado
+      AND NOT m.matricula_anulada
+      AND au.periodo_academico_id = 14
+      AND au.ciclo_id = 28
+), balotarios_alumno AS (
+    SELECT a.matricula_id, b.balotario_id, b.periodo_id, b.curso_id, b.formulario_url
+    FROM alumnos a
+             INNER JOIN academico.ac_balotario_aula ba ON ba.aula_id = a.aula_id
+             INNER JOIN academico.ac_balotario b ON b.balotario_id = ba.balotario_id AND b.periodo_academico_id = 14
+    UNION
+    SELECT a.matricula_id, b.balotario_id, b.periodo_id, b.curso_id, b.formulario_url
+    FROM alumnos a
+             INNER JOIN academico.ac_balotario_respuesta r ON r.matricula_id = a.matricula_id
+             INNER JOIN academico.ac_balotario b ON b.balotario_id = r.balotario_id AND b.periodo_academico_id = 14
+), cursos AS (
+    SELECT DISTINCT ON (s.matricula_id, s.periodo_id, s.curso_id)
+           s.matricula_id,
+           s.periodo_id,
+           s.balotario_id,
+           s.curso_id,
+           hc.curso_nombre,
+           s.formulario_url,
+           (r.balotario_respuesta_id IS NOT NULL) AS responde,
+           r.balotario_nota
+    FROM balotarios_alumno s
+             LEFT JOIN horario.ho_curso hc ON hc.curso_id = s.curso_id
+             LEFT JOIN academico.ac_balotario_respuesta r ON r.balotario_id = s.balotario_id AND r.matricula_id = s.matricula_id
+    ORDER BY s.matricula_id, s.periodo_id, s.curso_id, (r.balotario_respuesta_id IS NOT NULL) DESC, s.balotario_id
+), cursos_periodo AS (
+    SELECT matricula_id,
+           periodo_id,
+           JSONB_AGG(JSONB_BUILD_OBJECT('balotario_id', balotario_id,
+                                        'curso_id', curso_id,
+                                        'curso_nombre', curso_nombre,
+                                        'responde', responde,
+                                        'puntuacion', balotario_nota,
+                                        'formulario_url', formulario_url)
+                     ORDER BY curso_id) AS cursos
+    FROM cursos
+    GROUP BY matricula_id, periodo_id
+)
+SELECT a.matricula_id,
+       a.estudiante_id,
+       JSONB_AGG(JSONB_BUILD_OBJECT('periodo_id', p.periodo_id,
+                                    'periodo_nombre', cat.catalogo_siiaa_nombre,
+                                    'periodo_orden', p.periodo_orden,
+                                    'cursos', COALESCE(cp.cursos, '[]'::jsonb))
+                 ORDER BY p.periodo_orden) AS periodos
+FROM alumnos a
+         CROSS JOIN periodos p
+         INNER JOIN "general".gn_catalogo_siiaa cat ON cat.catalogo_siiaa_id = p.periodo_id
+         LEFT JOIN cursos_periodo cp ON cp.matricula_id = a.matricula_id AND cp.periodo_id = p.periodo_id
+GROUP BY a.matricula_id, a.estudiante_id
+ORDER BY a.matricula_id
+`
+
+// BalotarioByIDQuery es BalotarioQuery acotado a una sola matrícula, para la carga parcial
+// (upsert puntual) que dispara /api/v1/sync/balotarios/upsert.
+const BalotarioByIDQuery = `
+WITH periodos (periodo_id, periodo_orden) AS (
+    VALUES (3170, 1), (3174, 2), (3171, 3), (3175, 4), (3172, 5), (3176, 6), (3173, 7), (3177, 8)
+), alumnos AS (
+    SELECT m.matricula_id, m.estudiante_id, m.aula_id_asiste AS aula_id
+    FROM matricula.ma_matricula m
+             INNER JOIN academico.ac_aula au ON au.aula_id = m.aula_id_asiste
+    WHERE NOT m.matricula_retirado
+      AND NOT m.matricula_anulada
+      AND au.periodo_academico_id = 14
+      AND au.ciclo_id = 28
+      AND m.matricula_id = $1
+), balotarios_alumno AS (
+    SELECT a.matricula_id, b.balotario_id, b.periodo_id, b.curso_id, b.formulario_url
+    FROM alumnos a
+             INNER JOIN academico.ac_balotario_aula ba ON ba.aula_id = a.aula_id
+             INNER JOIN academico.ac_balotario b ON b.balotario_id = ba.balotario_id AND b.periodo_academico_id = 14
+    UNION
+    SELECT a.matricula_id, b.balotario_id, b.periodo_id, b.curso_id, b.formulario_url
+    FROM alumnos a
+             INNER JOIN academico.ac_balotario_respuesta r ON r.matricula_id = a.matricula_id
+             INNER JOIN academico.ac_balotario b ON b.balotario_id = r.balotario_id AND b.periodo_academico_id = 14
+), cursos AS (
+    SELECT DISTINCT ON (s.matricula_id, s.periodo_id, s.curso_id)
+           s.matricula_id,
+           s.periodo_id,
+           s.balotario_id,
+           s.curso_id,
+           hc.curso_nombre,
+           s.formulario_url,
+           (r.balotario_respuesta_id IS NOT NULL) AS responde,
+           r.balotario_nota
+    FROM balotarios_alumno s
+             LEFT JOIN horario.ho_curso hc ON hc.curso_id = s.curso_id
+             LEFT JOIN academico.ac_balotario_respuesta r ON r.balotario_id = s.balotario_id AND r.matricula_id = s.matricula_id
+    ORDER BY s.matricula_id, s.periodo_id, s.curso_id, (r.balotario_respuesta_id IS NOT NULL) DESC, s.balotario_id
+), cursos_periodo AS (
+    SELECT matricula_id,
+           periodo_id,
+           JSONB_AGG(JSONB_BUILD_OBJECT('balotario_id', balotario_id,
+                                        'curso_id', curso_id,
+                                        'curso_nombre', curso_nombre,
+                                        'responde', responde,
+                                        'puntuacion', balotario_nota,
+                                        'formulario_url', formulario_url)
+                     ORDER BY curso_id) AS cursos
+    FROM cursos
+    GROUP BY matricula_id, periodo_id
+)
+SELECT a.matricula_id,
+       a.estudiante_id,
+       JSONB_AGG(JSONB_BUILD_OBJECT('periodo_id', p.periodo_id,
+                                    'periodo_nombre', cat.catalogo_siiaa_nombre,
+                                    'periodo_orden', p.periodo_orden,
+                                    'cursos', COALESCE(cp.cursos, '[]'::jsonb))
+                 ORDER BY p.periodo_orden) AS periodos
+FROM alumnos a
+         CROSS JOIN periodos p
+         INNER JOIN "general".gn_catalogo_siiaa cat ON cat.catalogo_siiaa_id = p.periodo_id
+         LEFT JOIN cursos_periodo cp ON cp.matricula_id = a.matricula_id AND cp.periodo_id = p.periodo_id
+GROUP BY a.matricula_id, a.estudiante_id
+`

internal/infrastructure/database/postgres/balotario_reader.go

+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/jackc/pgx/v5"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+	"intranet-synchronizer/pkg/jsonutil"
+)
+
+const balotarioPeriodoAcademicoID = 14
+
+type BalotarioReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewBalotarioReader(pool *Pool, query string) repositories.SourceReader {
+	return &BalotarioReader{pool: pool, query: query}
+}
+
+func NewBalotarioRowReader(pool *Pool) repositories.SourceRowReader {
+	return &BalotarioReader{pool: pool, query: BalotarioByIDQuery}
+}
+
+func scanBalotarioRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var (
+		matriculaID  int
+		estudianteID int
+		periodosRaw  []byte
+	)
+	if err := row.Scan(&matriculaID, &estudianteID, &periodosRaw); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan balotario row: %w", err)
+	}
+
+	periodos, err := jsonutil.DecodeList[entities.BalotarioPeriodo](periodosRaw)
+	if err != nil {
+		return entities.SyncRow{}, fmt.Errorf("decode periodos for matricula %d: %w", matriculaID, err)
+	}
+	sortBalotarioPeriodos(periodos)
+
+	doc := &entities.Balotario{
+		MatriculaID:        matriculaID,
+		EstudianteID:       estudianteID,
+		PeriodoAcademicoID: balotarioPeriodoAcademicoID,
+		Periodos:           periodos,
+	}
+	return entities.SyncRow{ID: matriculaID, Doc: doc}, nil
+}
+
+func (f *BalotarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query balotarios: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanBalotarioRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate balotario rows: %w", err)
+	}
+	return result, nil
+}
+
+func (f *BalotarioReader) FetchByID(ctx context.Context, id int) (entities.SyncRow, bool, error) {
+	rows, err := f.pool.Pool.Query(ctx, BalotarioByIDQuery, id)
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("query balotario %d: %w", id, err)
+	}
+
+	row, err := pgx.CollectExactlyOneRow(rows, scanBalotarioRow)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return entities.SyncRow{}, false, nil
+	}
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("read balotario %d: %w", id, err)
+	}
+	return row, true, nil
+}
+
+func sortBalotarioPeriodos(periodos []entities.BalotarioPeriodo) {
+	sort.Slice(periodos, func(i, j int) bool {
+		return periodos[i].PeriodoOrden < periodos[j].PeriodoOrden
+	})
+	for i := range periodos {
+		if periodos[i].Cursos == nil {
+			periodos[i].Cursos = []entities.BalotarioCurso{}
+		}
+		cursos := periodos[i].Cursos
+		sort.Slice(cursos, func(a, b int) bool {
+			return cursos[a].CursoID < cursos[b].CursoID
+		})
+	}
+}

internal/infrastructure/database/postgres/client.go

@@ -29,7 +29,7 @@ CREATE TABLE IF NOT EXISTS sync_state (
 );
 
 INSERT INTO sync_state (id, last_synced_at)
-VALUES (1, '1970-01-01T00:00:00Z'), (2, '1970-01-01T00:00:00Z'), (3, '1970-01-01T00:00:00Z'), (4, '1970-01-01T00:00:00Z'), (5, '1970-01-01T00:00:00Z')
+VALUES (1, '1970-01-01T00:00:00Z'), (2, '1970-01-01T00:00:00Z'), (3, '1970-01-01T00:00:00Z'), (4, '1970-01-01T00:00:00Z'), (5, '1970-01-01T00:00:00Z'), (6, '1970-01-01T00:00:00Z')
 ON CONFLICT (id) DO NOTHING;
 `
 

internal/infrastructure/pipeline/builder.go

@@ -26,6 +26,8 @@ func readerFor(name string, pool *postgres.Pool, query string) (repositories.Sou
 		return postgres.NewUserReader(pool, query), nil
 	case "professors":
 		return postgres.NewProfessorReader(pool, query), nil
+	case "balotarios":
+		return postgres.NewBalotarioReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}
@@ -37,6 +39,8 @@ func rowReaderFor(name string, pool *postgres.Pool) repositories.SourceRowReader
 	switch name {
 	case "users":
 		return postgres.NewUserRowReader(pool)
+	case "balotarios":
+		return postgres.NewBalotarioRowReader(pool)
 	default:
 		return nil
 	}