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

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

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

internal/infrastructure/database/postgres/horario_query.go

@@ -75,3 +75,74 @@ FROM alumnos a
          LEFT JOIN bloques b ON b.aula_id = a.aula_id
 ORDER BY a.matricula_id
 `
+
+// HorarioByIDQuery es HorarioQuery acotado a una sola matrícula, para la carga parcial
+// (upsert puntual) que dispara /api/v1/sync/horario/upsert.
+const HorarioByIDQuery = `
+WITH alumnos AS (
+    SELECT m.matricula_id, m.estudiante_id,
+           au.aula_id, au.aula_nombre,
+           au.periodo_academico_id, au.ciclo_id,
+           au.nivel_id, au.grado_id, au.turno_id, au.seccion_id
+    FROM matricula.ma_matricula m
+             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
+), clases AS (
+    SELECT h.aula_id, h.dia, hp.bloque, hp.hora_inicio, hp.hora_termino,
+           hp.curso_id, c.curso_nombre, p.profesor_nombre
+    FROM horario.ho_horario h
+             JOIN horario.ho_horario_profesor hp ON hp.horario_id = h.horario_id
+             LEFT JOIN horario.ho_curso c ON c.curso_id = hp.curso_id
+             LEFT JOIN horario.ho_profesor p ON p.profesor_id = hp.profesor_id
+    WHERE hp.libre = '0'
+      AND hp.curso_id IS NOT NULL
+      AND hp.profesor_id IS NOT NULL
+      AND h.aula_id IN (SELECT DISTINCT aula_id FROM alumnos)
+), clases_bloque AS (
+    SELECT aula_id, bloque,
+           JSONB_AGG(JSONB_BUILD_OBJECT(
+                   'dia', dia,
+                   'hora_inicio', to_char(hora_inicio, 'HH24:MI'),
+                   'hora_termino', to_char(hora_termino, 'HH24:MI'),
+                   'curso_id', curso_id,
+                   'curso_nombre', curso_nombre,
+                   'profesor_nombre', profesor_nombre
+               ) ORDER BY dia) AS clases
+    FROM clases
+    GROUP BY aula_id, bloque
+), bloques AS (
+    SELECT aula_id,
+           JSONB_AGG(JSONB_BUILD_OBJECT('bloque', bloque, 'clases', clases) ORDER BY bloque) AS bloques
+    FROM clases_bloque
+    GROUP BY aula_id
+)
+SELECT a.matricula_id,
+       a.estudiante_id,
+       a.aula_id,
+       a.aula_nombre,
+       a.periodo_academico_id,
+       per.periodo_academico_nombre,
+       a.ciclo_id,
+       cic.catalogo_siiaa_nombre AS ciclo_nombre,
+       a.nivel_id,
+       niv.catalogo_siiaa_nombre AS nivel_nombre,
+       a.grado_id,
+       g.grado_sigla,
+       a.turno_id,
+       tur.catalogo_siiaa_nombre AS turno_nombre,
+       a.seccion_id,
+       sec.catalogo_siiaa_nombre AS seccion_nombre,
+       COALESCE(b.bloques, '[]'::jsonb) AS bloques
+FROM alumnos a
+         LEFT JOIN general.gn_periodo_academico per ON per.periodo_academico_id = a.periodo_academico_id
+         LEFT JOIN general.gn_catalogo_siiaa cic ON cic.catalogo_siiaa_id = a.ciclo_id
+         LEFT JOIN general.gn_catalogo_siiaa niv ON niv.catalogo_siiaa_id = a.nivel_id
+         LEFT JOIN academico.ac_grado g ON g.grado_id = a.grado_id
+         LEFT JOIN general.gn_catalogo_siiaa tur ON tur.catalogo_siiaa_id = a.turno_id
+         LEFT JOIN general.gn_catalogo_siiaa sec ON sec.catalogo_siiaa_id = a.seccion_id
+         LEFT JOIN bloques b ON b.aula_id = a.aula_id
+`

internal/infrastructure/database/postgres/horario_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"
@@ -19,15 +22,11 @@ func NewHorarioReader(pool *Pool, query string) repositories.SourceReader {
 	return &HorarioReader{pool: pool, query: query}
 }
 
-func (f *HorarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
-	rows, err := f.pool.Pool.Query(ctx, f.query)
-	if err != nil {
-		return nil, fmt.Errorf("query horario: %w", err)
-	}
-	defer rows.Close()
+func NewHorarioRowReader(pool *Pool) repositories.SourceRowReader {
+	return &HorarioReader{pool: pool, query: HorarioByIDQuery}
+}
 
-	var result []entities.SyncRow
-	for rows.Next() {
+func scanHorarioRow(row pgx.CollectableRow) (entities.SyncRow, error) {
 	var (
 		matriculaID            int
 		estudianteID           int
@@ -47,16 +46,16 @@ func (f *HorarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error
 		seccionNombre          *string
 		bloquesRaw             []byte
 	)
-		if err := rows.Scan(&matriculaID, &estudianteID, &aulaID, &aulaNombre,
+	if err := row.Scan(&matriculaID, &estudianteID, &aulaID, &aulaNombre,
 		&periodoAcademicoID, &periodoAcademicoNombre, &cicloID, &cicloNombre,
 		&nivelID, &nivelNombre, &gradoID, &gradoSigla,
 		&turnoID, &turnoNombre, &seccionID, &seccionNombre, &bloquesRaw); err != nil {
-			return nil, fmt.Errorf("scan horario row: %w", err)
+		return entities.SyncRow{}, fmt.Errorf("scan horario row: %w", err)
 	}
 
 	bloques, err := jsonutil.DecodeList[entities.HorarioBloque](bloquesRaw)
 	if err != nil {
-			return nil, fmt.Errorf("decode bloques for matricula %d: %w", matriculaID, err)
+		return entities.SyncRow{}, fmt.Errorf("decode bloques for matricula %d: %w", matriculaID, err)
 	}
 	sortHorarioBloques(bloques)
 
@@ -81,14 +80,38 @@ func (f *HorarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error
 		},
 		Bloques: bloques,
 	}
-		result = append(result, entities.SyncRow{ID: matriculaID, Doc: doc})
+	return entities.SyncRow{ID: matriculaID, Doc: doc}, nil
+}
+
+func (f *HorarioReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query horario: %w", err)
 	}
-	if err := rows.Err(); err != nil {
+
+	result, err := pgx.CollectRows(rows, scanHorarioRow)
+	if err != nil {
 		return nil, fmt.Errorf("iterate horario rows: %w", err)
 	}
 	return result, nil
 }
 
+func (f *HorarioReader) FetchByID(ctx context.Context, id int) (entities.SyncRow, bool, error) {
+	rows, err := f.pool.Pool.Query(ctx, HorarioByIDQuery, id)
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("query horario %d: %w", id, err)
+	}
+
+	row, err := pgx.CollectExactlyOneRow(rows, scanHorarioRow)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return entities.SyncRow{}, false, nil
+	}
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("read horario %d: %w", id, err)
+	}
+	return row, true, nil
+}
+
 func sortHorarioBloques(bloques []entities.HorarioBloque) {
 	sort.Slice(bloques, func(i, j int) bool {
 		return bloques[i].Bloque < bloques[j].Bloque

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 "horario":
+		return postgres.NewHorarioRowReader(pool)
 	default:
 		return nil
 	}