feat: pipeline de sincronización de horario hacia c_horario

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

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

@@ -52,4 +52,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "professor_id",
 		StateID:                5,
 	},
+	{
+		Name:                   "horario",
+		PGQuery:                postgres.HorarioQuery,
+		MongoCollection:        "c_horario",
+		MongoDeletedCollection: "deleted_c_horario",
+		IDField:                "matricula_id",
+		StateID:                6,
+	},
 }

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/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
+`

internal/infrastructure/database/postgres/horario_reader.go

+package postgres
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	"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 (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()
+
+	var result []entities.SyncRow
+	for rows.Next() {
+		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 := rows.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)
+		}
+
+		bloques, err := jsonutil.DecodeList[entities.HorarioBloque](bloquesRaw)
+		if err != nil {
+			return nil, 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,
+		}
+		result = append(result, entities.SyncRow{ID: matriculaID, Doc: doc})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate horario rows: %w", err)
+	}
+	return result, 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

@@ -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 "horario":
+		return postgres.NewHorarioReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}