merge: integrar pipeline de horario

# Conflicts:
#	internal/infrastructure/config/pipelines.go
#	internal/infrastructure/database/postgres/client.go
#	internal/infrastructure/pipeline/builder.go

internal/domain/entities/horario.go

+package entities
+
+import "time"
+
+type HorarioClase struct {
+	Dia            int    `bson:"dia" json:"dia"`
+	HoraInicio     string `bson:"hora_inicio" json:"hora_inicio"`
+	HoraTermino    string `bson:"hora_termino" json:"hora_termino"`
+	CursoID        int    `bson:"curso_id" json:"curso_id"`
+	CursoNombre    string `bson:"curso_nombre" json:"curso_nombre"`
+	ProfesorNombre string `bson:"profesor_nombre" json:"profesor_nombre"`
+}
+
+type HorarioBloque struct {
+	Bloque int            `bson:"bloque" json:"bloque"`
+	Clases []HorarioClase `bson:"clases" json:"clases"`
+}
+
+type HorarioAula struct {
+	AulaID                 int    `bson:"aula_id" json:"aula_id"`
+	AulaNombre             string `bson:"aula_nombre" json:"aula_nombre"`
+	PeriodoAcademicoID     int    `bson:"periodo_academico_id" json:"periodo_academico_id"`
+	PeriodoAcademicoNombre string `bson:"periodo_academico_nombre" json:"periodo_academico_nombre"`
+	CicloID                int    `bson:"ciclo_id" json:"ciclo_id"`
+	CicloNombre            string `bson:"ciclo_nombre" json:"ciclo_nombre"`
+	NivelID                int    `bson:"nivel_id" json:"nivel_id"`
+	NivelNombre            string `bson:"nivel_nombre" json:"nivel_nombre"`
+	GradoID                int    `bson:"grado_id" json:"grado_id"`
+	GradoSigla             string `bson:"grado_sigla" json:"grado_sigla"`
+	TurnoID                int    `bson:"turno_id" json:"turno_id"`
+	TurnoNombre            string `bson:"turno_nombre" json:"turno_nombre"`
+	SeccionID              int    `bson:"seccion_id" json:"seccion_id"`
+	SeccionNombre          string `bson:"seccion_nombre" json:"seccion_nombre"`
+}
+
+type Horario struct {
+	MatriculaID  int             `bson:"matricula_id"`
+	EstudianteID int             `bson:"estudiante_id"`
+	Aula         HorarioAula     `bson:"aula"`
+	Bloques      []HorarioBloque `bson:"bloques"`
+	CreatedAt    time.Time       `bson:"created_at"`
+	UpdatedAt    time.Time       `bson:"updated_at"`
+	RowHash      string          `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (r *Horario) SetUpdatedAt(t time.Time) {
+	r.UpdatedAt = t
+}
+
+func (r *Horario) SetRowHash(h string) {
+	r.RowHash = h
+}
+
+func (r *Horario) SetCreatedAt(t time.Time) {
+	r.CreatedAt = t
+}

internal/infrastructure/config/pipelines.go

@@ -68,4 +68,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "attendance_id",
 		StateID:                7,
 	},
+	{
+		Name:                   "horario",
+		PGQuery:                postgres.HorarioQuery,
+		MongoCollection:        "c_horario",
+		MongoDeletedCollection: "deleted_c_horario",
+		IDField:                "matricula_id",
+		StateID:                8,
+	},
 }

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'), (6, '1970-01-01T00:00:00Z'), (7, '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'), (7, '1970-01-01T00:00:00Z'), (8, '1970-01-01T00:00:00Z')
 ON CONFLICT (id) DO NOTHING;
 `
 

internal/infrastructure/database/postgres/horario_query.go

+package postgres
+
+// HorarioQuery arma un documento por matrícula activa del período 14 / ciclo 28 con el horario
+// semanal de su aula: bloques (1-8) con las clases de cada día (1=lunes ... 6=sábado).
+// El horario es por aula, no por alumno, por lo que se calcula una sola vez por aula
+// (subconsulta `clases`/`bloques` filtrada a las aulas con matrícula activa) y se repite
+// para cada alumno de esa aula.
+// Se excluyen los bloques con `libre = '1'` (hora libre, sin curso/profesor asignado) y los
+// que, sin estar marcados libres, tienen curso_id o profesor_id nulos (huecos de la data real).
+const HorarioQuery = `
+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
+), 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
+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

+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"
+)
+
+type HorarioReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewHorarioReader(pool *Pool, query string) repositories.SourceReader {
+	return &HorarioReader{pool: pool, query: query}
+}
+
+func NewHorarioRowReader(pool *Pool) repositories.SourceRowReader {
+	return &HorarioReader{pool: pool, query: HorarioByIDQuery}
+}
+
+func scanHorarioRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var (
+		matriculaID            int
+		estudianteID           int
+		aulaID                 int
+		aulaNombre             string
+		periodoAcademicoID     int
+		periodoAcademicoNombre *string
+		cicloID                int
+		cicloNombre            *string
+		nivelID                int
+		nivelNombre            *string
+		gradoID                int
+		gradoSigla             *string
+		turnoID                int
+		turnoNombre            *string
+		seccionID              int
+		seccionNombre          *string
+		bloquesRaw             []byte
+	)
+	if err := row.Scan(&matriculaID, &estudianteID, &aulaID, &aulaNombre,
+		&periodoAcademicoID, &periodoAcademicoNombre, &cicloID, &cicloNombre,
+		&nivelID, &nivelNombre, &gradoID, &gradoSigla,
+		&turnoID, &turnoNombre, &seccionID, &seccionNombre, &bloquesRaw); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan horario row: %w", err)
+	}
+
+	bloques, err := jsonutil.DecodeList[entities.HorarioBloque](bloquesRaw)
+	if err != nil {
+		return entities.SyncRow{}, fmt.Errorf("decode bloques for matricula %d: %w", matriculaID, err)
+	}
+	sortHorarioBloques(bloques)
+
+	doc := &entities.Horario{
+		MatriculaID:  matriculaID,
+		EstudianteID: estudianteID,
+		Aula: entities.HorarioAula{
+			AulaID:                 aulaID,
+			AulaNombre:             aulaNombre,
+			PeriodoAcademicoID:     periodoAcademicoID,
+			PeriodoAcademicoNombre: derefString(periodoAcademicoNombre),
+			CicloID:                cicloID,
+			CicloNombre:            derefString(cicloNombre),
+			NivelID:                nivelID,
+			NivelNombre:            derefString(nivelNombre),
+			GradoID:                gradoID,
+			GradoSigla:             derefString(gradoSigla),
+			TurnoID:                turnoID,
+			TurnoNombre:            derefString(turnoNombre),
+			SeccionID:              seccionID,
+			SeccionNombre:          derefString(seccionNombre),
+		},
+		Bloques: bloques,
+	}
+	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)
+	}
+
+	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
+	})
+	for i := range bloques {
+		if bloques[i].Clases == nil {
+			bloques[i].Clases = []entities.HorarioClase{}
+		}
+		clases := bloques[i].Clases
+		sort.Slice(clases, func(a, b int) bool {
+			return clases[a].Dia < clases[b].Dia
+		})
+	}
+}
+
+func derefString(s *string) string {
+	if s == nil {
+		return ""
+	}
+	return *s
+}

internal/infrastructure/pipeline/builder.go

@@ -30,6 +30,8 @@ func readerFor(name string, pool *postgres.Pool, query string) (repositories.Sou
 		return postgres.NewBalotarioReader(pool, query), nil
 	case "attendance":
 		return postgres.NewAttendanceReader(pool, query), nil
+	case "horario":
+		return postgres.NewHorarioReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}
@@ -43,6 +45,8 @@ func rowReaderFor(name string, pool *postgres.Pool) repositories.SourceRowReader
 		return postgres.NewUserRowReader(pool)
 	case "balotarios":
 		return postgres.NewBalotarioRowReader(pool)
+	case "horario":
+		return postgres.NewHorarioRowReader(pool)
 	default:
 		return nil
 	}