merge: integrar pipeline de asistencia

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

cmd/server/main.go

@@ -54,10 +54,11 @@ func main() {
 	}
 
 	r := router.New(router.Config{
-		SyncUseCases:   useCases,
-		UpsertUseCases: pipeline.BuildUpserters(cfg, pool, mongoClient),
-		PGPing:         func(pingCtx context.Context) error { return pool.Ping(pingCtx) },
-		MongoPing:      func(pingCtx context.Context) error { return mongoClient.Ping(pingCtx, nil) },
+		SyncUseCases:        useCases,
+		UpsertUseCases:      pipeline.BuildUpserters(cfg, pool, mongoClient),
+		UpsertGroupUseCases: pipeline.BuildGroupUpserters(cfg, pool, mongoClient),
+		PGPing:              func(pingCtx context.Context) error { return pool.Ping(pingCtx) },
+		MongoPing:           func(pingCtx context.Context) error { return mongoClient.Ping(pingCtx, nil) },
 	})
 
 	srv := &http.Server{Addr: ":" + cfg.Port, Handler: r}

internal/domain/entities/attendance.go

+package entities
+
+import "time"
+
+type AttendanceRecord struct {
+	AsistenciaID        *int64  `bson:"asistencia_id" json:"asistencia_id"`
+	Fecha               string  `bson:"fecha" json:"fecha"`
+	EsFeriado           bool    `bson:"es_feriado" json:"es_feriado"`
+	HoraIngreso         *string `bson:"hora_ingreso" json:"hora_ingreso"`
+	Tipo                string  `bson:"tipo" json:"tipo"`
+	Justificado         *bool   `bson:"justificado" json:"justificado"`
+	MotivoJustificacion *string `bson:"motivo_justificacion" json:"motivo_justificacion"`
+}
+
+type AttendanceSummary struct {
+	DiasCalendario     int `bson:"dias_calendario" json:"dias_calendario"`
+	DiasHabiles        int `bson:"dias_habiles" json:"dias_habiles"`
+	Presentes          int `bson:"presentes" json:"presentes"`
+	Tardanzas          int `bson:"tardanzas" json:"tardanzas"`
+	Faltas             int `bson:"faltas" json:"faltas"`
+	FaltasJustificadas int `bson:"faltas_justificadas" json:"faltas_justificadas"`
+	Feriados           int `bson:"feriados" json:"feriados"`
+}
+
+type Attendance struct {
+	AttendanceID       int                `bson:"attendance_id"`
+	MatriculaID        int                `bson:"matricula_id"`
+	EstudianteID       int                `bson:"estudiante_id"`
+	Mes                int                `bson:"mes"`
+	MesDescripcion     string             `bson:"mes_descripcion"`
+	PeriodoAcademicoID int                `bson:"periodo_academico_id"`
+	CicloID            int                `bson:"ciclo_id"`
+	Registros          []AttendanceRecord `bson:"registros"`
+	ResumenMes         AttendanceSummary  `bson:"resumen_mes"`
+	CreatedAt          time.Time          `bson:"created_at"`
+	UpdatedAt          time.Time          `bson:"updated_at"`
+	RowHash            string             `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (r *Attendance) SetUpdatedAt(t time.Time) {
+	r.UpdatedAt = t
+}
+
+func (r *Attendance) SetRowHash(h string) {
+	r.RowHash = h
+}
+
+func (r *Attendance) SetCreatedAt(t time.Time) {
+	r.CreatedAt = t
+}

internal/domain/repositories/sync_repository.go

@@ -15,6 +15,12 @@ type SourceRowReader interface {
 	FetchByID(ctx context.Context, id int) (entities.SyncRow, bool, error)
 }
 
+// SourceGroupReader lo implementan los módulos cuyo id de negocio agrupa varios documentos,
+// como asistencia, donde una matrícula produce un documento por mes.
+type SourceGroupReader interface {
+	FetchByGroupID(ctx context.Context, groupID int) ([]entities.SyncRow, error)
+}
+
 type DocumentUpserter interface {
 	BulkUpsert(ctx context.Context, rows []entities.SyncRow) (created, updated int, err error)
 }

internal/infrastructure/config/pipelines.go

@@ -60,4 +60,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "matricula_id",
 		StateID:                6,
 	},
+	{
+		Name:                   "attendance",
+		PGQuery:                postgres.AttendanceQuery,
+		MongoCollection:        "c_attendance",
+		MongoDeletedCollection: "deleted_c_attendance",
+		IDField:                "attendance_id",
+		StateID:                7,
+	},
 }

internal/infrastructure/database/postgres/attendance_query.go

+package postgres
+
+import "fmt"
+
+// attendanceQueryTemplate arma un documento por (matricula, mes) del año en curso.
+// Recorre el calendario y hace LEFT JOIN contra ac_asistencia, así que todo día del año
+// aparece: con datos, feriado, o sin fila cuando no se tomó asistencia.
+// periodo_academico_id y ciclo_id quedan fijos a propósito, igual que en los demás módulos.
+const attendanceQueryTemplate = `
+WITH alumnos AS (
+    SELECT m.matricula_id, m.estudiante_id, au.periodo_academico_id, au.ciclo_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
+       %s
+), calendario AS (
+    SELECT c.calendario_fecha AS fecha, c.calendario_es_feriado AS es_feriado
+      FROM configuracion.co_calendario c
+     WHERE EXTRACT(YEAR FROM c.calendario_fecha) = EXTRACT(YEAR FROM CURRENT_DATE)
+       AND c.calendario_fecha <= CURRENT_DATE
+), dias AS (
+    SELECT al.matricula_id, al.estudiante_id, al.periodo_academico_id, al.ciclo_id,
+           cal.fecha, cal.es_feriado,
+           a.asistencia_id, a.hora_ingreso, a.tipo_asistencia_id,
+           a.justifico, a.justifico_observacion
+      FROM alumnos al
+      CROSS JOIN calendario cal
+      LEFT JOIN academico.ac_asistencia a
+             ON a.matricula_id = al.matricula_id
+            AND a.asistencia_fecha = cal.fecha
+)
+SELECT d.matricula_id,
+       d.estudiante_id,
+       EXTRACT(MONTH FROM d.fecha)::int AS mes,
+       CASE EXTRACT(MONTH FROM d.fecha)::int
+           WHEN 1 THEN 'Enero'   WHEN 2 THEN 'Febrero'   WHEN 3  THEN 'Marzo'
+           WHEN 4 THEN 'Abril'   WHEN 5 THEN 'Mayo'      WHEN 6  THEN 'Junio'
+           WHEN 7 THEN 'Julio'   WHEN 8 THEN 'Agosto'    WHEN 9  THEN 'Septiembre'
+           WHEN 10 THEN 'Octubre' WHEN 11 THEN 'Noviembre' WHEN 12 THEN 'Diciembre'
+       END AS mes_descripcion,
+       d.periodo_academico_id,
+       d.ciclo_id,
+       JSON_AGG(
+           JSON_BUILD_OBJECT(
+               'asistencia_id', d.asistencia_id,
+               'fecha', to_char(d.fecha, 'YYYY-MM-DD'),
+               'es_feriado', d.es_feriado,
+               'hora_ingreso', d.hora_ingreso,
+               'tipo', COALESCE(CASE d.tipo_asistencia_id
+                                    WHEN 1 THEN 'asistencia'
+                                    WHEN 2 THEN 'tardanza'
+                                    WHEN 3 THEN 'falta'
+                                    WHEN 4 THEN 'no_corresponde'
+                                END, ''),
+               'justificado', COALESCE(d.justifico, false),
+               'motivo_justificacion', d.justifico_observacion
+           ) ORDER BY d.fecha
+       ) AS registros,
+       JSON_BUILD_OBJECT(
+           'dias_calendario', COUNT(*),
+           'dias_habiles', COUNT(*) FILTER (WHERE d.tipo_asistencia_id IS NOT NULL),
+           'presentes', COUNT(*) FILTER (WHERE d.tipo_asistencia_id = 1),
+           'tardanzas', COUNT(*) FILTER (WHERE d.tipo_asistencia_id = 2),
+           'faltas', COUNT(*) FILTER (WHERE d.tipo_asistencia_id = 3),
+           'faltas_justificadas', COUNT(*) FILTER (WHERE d.tipo_asistencia_id = 3 AND d.justifico),
+           'feriados', COUNT(*) FILTER (WHERE d.es_feriado)
+       ) AS resumen_mes
+  FROM dias d
+ GROUP BY d.matricula_id, d.estudiante_id, d.periodo_academico_id, d.ciclo_id,
+          EXTRACT(MONTH FROM d.fecha)
+ ORDER BY d.matricula_id, EXTRACT(MONTH FROM d.fecha)`
+
+// AttendanceQuery lee todas las matrículas activas del período vigente.
+var AttendanceQuery = fmt.Sprintf(attendanceQueryTemplate, "")
+
+// AttendanceByMatriculaQuery lee los meses de una sola matrícula, para el upsert bajo demanda.
+var AttendanceByMatriculaQuery = fmt.Sprintf(attendanceQueryTemplate, "AND m.matricula_id = $1")

internal/infrastructure/database/postgres/attendance_reader.go

+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+)
+
+// attendanceDocID arma el id sintético del documento, porque la clave natural del módulo es
+// (matricula_id, mes) y el pipeline solo admite un id entero.
+func attendanceDocID(matriculaID, mes int) int {
+	return matriculaID*100 + mes
+}
+
+type AttendanceReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewAttendanceReader(pool *Pool, query string) repositories.SourceReader {
+	return &AttendanceReader{pool: pool, query: query}
+}
+
+func NewAttendanceGroupReader(pool *Pool) repositories.SourceGroupReader {
+	return &AttendanceReader{pool: pool, query: AttendanceByMatriculaQuery}
+}
+
+func scanAttendanceRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var (
+		matriculaID        int
+		estudianteID       int
+		mes                int
+		mesDescripcion     string
+		periodoAcademicoID int
+		cicloID            int
+		registrosJSON      []byte
+		resumenJSON        []byte
+	)
+	if err := row.Scan(&matriculaID, &estudianteID, &mes, &mesDescripcion,
+		&periodoAcademicoID, &cicloID, &registrosJSON, &resumenJSON); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan attendance row: %w", err)
+	}
+
+	var registros []entities.AttendanceRecord
+	if err := json.Unmarshal(registrosJSON, &registros); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("unmarshal registros matricula %d mes %d: %w", matriculaID, mes, err)
+	}
+
+	var resumen entities.AttendanceSummary
+	if err := json.Unmarshal(resumenJSON, &resumen); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("unmarshal resumen matricula %d mes %d: %w", matriculaID, mes, err)
+	}
+
+	doc := &entities.Attendance{
+		AttendanceID:       attendanceDocID(matriculaID, mes),
+		MatriculaID:        matriculaID,
+		EstudianteID:       estudianteID,
+		Mes:                mes,
+		MesDescripcion:     mesDescripcion,
+		PeriodoAcademicoID: periodoAcademicoID,
+		CicloID:            cicloID,
+		Registros:          registros,
+		ResumenMes:         resumen,
+	}
+	return entities.SyncRow{ID: attendanceDocID(matriculaID, mes), Doc: doc}, nil
+}
+
+func (f *AttendanceReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query attendance: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanAttendanceRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate attendance rows: %w", err)
+	}
+	return result, nil
+}
+
+// FetchByGroupID devuelve todos los meses de una matrícula, porque un alumno se sincroniza
+// completo en una sola llamada aunque cada mes sea un documento aparte.
+func (f *AttendanceReader) FetchByGroupID(ctx context.Context, matriculaID int) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, AttendanceByMatriculaQuery, matriculaID)
+	if err != nil {
+		return nil, fmt.Errorf("query attendance matricula %d: %w", matriculaID, err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanAttendanceRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate attendance rows matricula %d: %w", matriculaID, err)
+	}
+	return result, nil
+}

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')
+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')
 ON CONFLICT (id) DO NOTHING;
 `
 

internal/infrastructure/http/handlers/upsert_group_handler.go

+package handlers
+
+import (
+	"log"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	"intranet-synchronizer/internal/usecase"
+)
+
+type UpsertGroupHandler struct {
+	name    string
+	idField string
+	upsert  usecase.UpsertGroupUseCase
+}
+
+func NewUpsertGroupHandler(name, idField string, upsert usecase.UpsertGroupUseCase) *UpsertGroupHandler {
+	return &UpsertGroupHandler{name: name, idField: idField, upsert: upsert}
+}
+
+type upsertGroupRequest struct {
+	ID int `json:"id" binding:"required"`
+}
+
+func (h *UpsertGroupHandler) Upsert(c *gin.Context) {
+	var req upsertGroupRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		log.Printf("[%s upsert-group] payload inválido: %v", h.name, err)
+		c.JSON(http.StatusBadRequest, gin.H{"status": false, "message": "id es obligatorio"})
+		return
+	}
+
+	result, err := h.upsert.Upsert(c.Request.Context(), req.ID)
+	if err != nil {
+		log.Printf("[%s upsert-group] %s=%d error: %v", h.name, h.idField, req.ID, err)
+		c.JSON(http.StatusInternalServerError, gin.H{"status": false, "message": err.Error()})
+		return
+	}
+
+	log.Printf("[%s upsert-group] %s=%d created=%d updated=%d unchanged=%d",
+		h.name, h.idField, req.ID, result.Created, result.Updated, result.Unchanged)
+
+	if result.Total() == 0 {
+		c.JSON(http.StatusNotFound, gin.H{
+			"status":  false,
+			"message": "no existe el registro en el origen",
+			"module":  h.name,
+			"id":      req.ID,
+		})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"status":    true,
+		"module":    h.name,
+		"id":        req.ID,
+		"documents": result.Total(),
+		"created":   result.Created,
+		"updated":   result.Updated,
+		"unchanged": result.Unchanged,
+	})
+}

internal/infrastructure/http/router/router.go

@@ -5,15 +5,17 @@ import (
 
 	"github.com/gin-gonic/gin"
 
+	"intranet-synchronizer/internal/infrastructure/config"
 	"intranet-synchronizer/internal/infrastructure/http/handlers"
 	"intranet-synchronizer/internal/usecase"
 )
 
 type Config struct {
-	SyncUseCases   map[string]usecase.RunSyncPipelineUseCase
-	UpsertUseCases map[string]usecase.UpsertDocumentUseCase
-	PGPing         func(context.Context) error
-	MongoPing      func(context.Context) error
+	SyncUseCases        map[string]usecase.RunSyncPipelineUseCase
+	UpsertUseCases      map[string]usecase.UpsertDocumentUseCase
+	UpsertGroupUseCases map[string]usecase.UpsertGroupUseCase
+	PGPing              func(context.Context) error
+	MongoPing           func(context.Context) error
 }
 
 func New(cfg Config) *gin.Engine {
@@ -34,5 +36,19 @@ func New(cfg Config) *gin.Engine {
 		h := handlers.NewUpsertHandler(name, uc)
 		v1.POST("/sync/"+name+"/upsert", h.Upsert)
 	}
+
+	for name, uc := range cfg.UpsertGroupUseCases {
+		h := handlers.NewUpsertGroupHandler(name, idFieldFor(name), uc)
+		v1.POST("/sync/"+name+"/upsert-group", h.Upsert)
+	}
 	return r
 }
+
+func idFieldFor(name string) string {
+	for _, p := range config.Pipelines {
+		if p.Name == name {
+			return p.IDField
+		}
+	}
+	return "id"
+}

internal/infrastructure/pipeline/builder.go

@@ -28,6 +28,8 @@ func readerFor(name string, pool *postgres.Pool, query string) (repositories.Sou
 		return postgres.NewProfessorReader(pool, query), nil
 	case "balotarios":
 		return postgres.NewBalotarioReader(pool, query), nil
+	case "attendance":
+		return postgres.NewAttendanceReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}
@@ -46,6 +48,38 @@ func rowReaderFor(name string, pool *postgres.Pool) repositories.SourceRowReader
 	}
 }
 
+// groupReaderFor devuelve el lector por id de grupo de los módulos cuyo id de negocio agrupa
+// varios documentos. Los módulos sin lector de grupo no exponen el endpoint /upsert-group.
+func groupReaderFor(name string, pool *postgres.Pool) repositories.SourceGroupReader {
+	switch name {
+	case "attendance":
+		return postgres.NewAttendanceGroupReader(pool)
+	default:
+		return nil
+	}
+}
+
+// BuildGroupUpserters arma un caso de uso de upsert por grupo para cada pipeline que lo soporte.
+func BuildGroupUpserters(
+	cfg config.Config,
+	pool *postgres.Pool,
+	mongoClient *mongodriver.Client,
+) map[string]usecase.UpsertGroupUseCase {
+	useCases := make(map[string]usecase.UpsertGroupUseCase)
+
+	for _, p := range config.Pipelines {
+		reader := groupReaderFor(p.Name, pool)
+		if reader == nil {
+			continue
+		}
+
+		docRepo := repository.NewDocumentRepository(mongoClient.Database(cfg.MongoDB).Collection(p.MongoCollection), p.IDField)
+		useCases[p.Name] = usecase.NewUpsertGroupUseCase(reader, docRepo, docRepo, time.Now)
+	}
+
+	return useCases
+}
+
 // BuildUpserters arma un caso de uso de upsert puntual por cada pipeline que tenga lector por id.
 func BuildUpserters(
 	cfg config.Config,

internal/usecase/upsert_group.go

+package usecase
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+)
+
+type UpsertGroupResult struct {
+	Created   int
+	Updated   int
+	Unchanged int
+}
+
+func (r UpsertGroupResult) Total() int {
+	return r.Created + r.Updated + r.Unchanged
+}
+
+// UpsertGroupUseCase sincroniza bajo demanda todos los documentos que cuelgan de un id de
+// negocio. Existe para módulos como asistencia, donde una matrícula produce un documento por
+// mes y sincronizarla mes a mes obligaría a una llamada por cada uno.
+type UpsertGroupUseCase interface {
+	Upsert(ctx context.Context, groupID int) (UpsertGroupResult, error)
+}
+
+type upsertGroup struct {
+	reader   repositories.SourceGroupReader
+	upserter repositories.DocumentUpserter
+	fetcher  repositories.DocumentFetcher
+	now      func() time.Time
+}
+
+func NewUpsertGroupUseCase(
+	reader repositories.SourceGroupReader,
+	upserter repositories.DocumentUpserter,
+	fetcher repositories.DocumentFetcher,
+	now func() time.Time,
+) UpsertGroupUseCase {
+	return &upsertGroup{reader: reader, upserter: upserter, fetcher: fetcher, now: now}
+}
+
+func (s *upsertGroup) Upsert(ctx context.Context, groupID int) (UpsertGroupResult, error) {
+	rows, err := s.reader.FetchByGroupID(ctx, groupID)
+	if err != nil {
+		return UpsertGroupResult{}, fmt.Errorf("fetch by group id: %w", err)
+	}
+	if len(rows) == 0 {
+		return UpsertGroupResult{}, nil
+	}
+
+	ids := make([]int, 0, len(rows))
+	for _, row := range rows {
+		ids = append(ids, row.ID)
+	}
+
+	existing, err := s.fetcher.FindByIDs(ctx, ids)
+	if err != nil {
+		return UpsertGroupResult{}, fmt.Errorf("fetch current docs: %w", err)
+	}
+
+	runAt := s.now().UTC()
+	var result UpsertGroupResult
+	changed := make([]entities.SyncRow, 0, len(rows))
+
+	for _, row := range rows {
+		hash, err := computeRowHash(row.Doc)
+		if err != nil {
+			return UpsertGroupResult{}, err
+		}
+
+		current, exists := existing[row.ID]
+		if exists {
+			if storedHash, _ := current["row_hash"].(string); storedHash == hash {
+				result.Unchanged++
+				continue
+			}
+			result.Updated++
+		} else {
+			result.Created++
+		}
+
+		stampUpdatedAt(row.Doc, runAt)
+		stampRowHash(row.Doc, hash)
+		stampCreatedAt(row.Doc, runAt)
+		changed = append(changed, row)
+	}
+
+	if len(changed) == 0 {
+		return result, nil
+	}
+
+	if _, _, err := s.upserter.BulkUpsert(ctx, changed); err != nil {
+		return UpsertGroupResult{}, fmt.Errorf("upsert: %w", err)
+	}
+	return result, nil
+}