feat: agregar carga parcial (upsert por matrícula) al pipeline de balotarios

BalotarioByIDQuery acota BalotarioQuery a una sola matrícula;
BalotarioReader gana FetchByID reutilizando el scan de FetchAll, y
rowReaderFor lo expone como /api/v1/sync/balotarios/upsert (mismo
patrón que users y horario).

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

internal/infrastructure/database/postgres/balotario_query.go

@@ -62,3 +62,68 @@ FROM alumnos a
 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

@@ -2,9 +2,12 @@ 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"
@@ -21,27 +24,23 @@ func NewBalotarioReader(pool *Pool, query string) repositories.SourceReader {
 	return &BalotarioReader{pool: pool, query: query}
 }
 
-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)
-	}
-	defer rows.Close()
+func NewBalotarioRowReader(pool *Pool) repositories.SourceRowReader {
+	return &BalotarioReader{pool: pool, query: BalotarioByIDQuery}
+}
 
-	var result []entities.SyncRow
-	for rows.Next() {
+func scanBalotarioRow(row pgx.CollectableRow) (entities.SyncRow, error) {
 	var (
 		matriculaID  int
 		estudianteID int
 		periodosRaw  []byte
 	)
-		if err := rows.Scan(&matriculaID, &estudianteID, &periodosRaw); err != nil {
-			return nil, fmt.Errorf("scan balotario row: %w", err)
+	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 nil, fmt.Errorf("decode periodos for matricula %d: %w", matriculaID, err)
+		return entities.SyncRow{}, fmt.Errorf("decode periodos for matricula %d: %w", matriculaID, err)
 	}
 	sortBalotarioPeriodos(periodos)
 
@@ -51,14 +50,38 @@ func (f *BalotarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, err
 		PeriodoAcademicoID: balotarioPeriodoAcademicoID,
 		Periodos:           periodos,
 	}
-		result = append(result, entities.SyncRow{ID: matriculaID, Doc: doc})
+	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)
 	}
-	if err := rows.Err(); err != nil {
+
+	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

internal/infrastructure/pipeline/builder.go

@@ -39,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
 	}