feat: pipeline de sincronización de mensajes hacia c_mensajes

Un documento por matrícula destinataria con sus mensajes activos
(mensaje_estado = 1) enviados por directores desde Trismegisto vía
intranet.mensaje / intranet.mensaje_destinatario. Incluye FetchByID
desde el día uno: es el camino principal de entrega (Trismegisto
dispara el upsert al enviar), no solo una carga parcial de respaldo.

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

internal/domain/entities/mensaje.go

+package entities
+
+import "time"
+
+type MensajeItem struct {
+	MensajeID       int    `bson:"mensaje_id" json:"mensaje_id"`
+	Titulo          string `bson:"titulo" json:"titulo"`
+	Cuerpo          string `bson:"cuerpo" json:"cuerpo"`
+	ImagenURL       string `bson:"imagen_url" json:"imagen_url"`
+	Fecha           string `bson:"fecha" json:"fecha"`
+	RemitenteNombre string `bson:"remitente_nombre" json:"remitente_nombre"`
+	SedeNombre      string `bson:"sede_nombre" json:"sede_nombre"`
+}
+
+type Mensaje struct {
+	MatriculaID int           `bson:"matricula_id"`
+	Mensajes    []MensajeItem `bson:"mensajes"`
+	CreatedAt   time.Time     `bson:"created_at"`
+	UpdatedAt   time.Time     `bson:"updated_at"`
+	RowHash     string        `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (m *Mensaje) SetUpdatedAt(t time.Time) {
+	m.UpdatedAt = t
+}
+
+func (m *Mensaje) SetRowHash(h string) {
+	m.RowHash = h
+}
+
+func (m *Mensaje) SetCreatedAt(t time.Time) {
+	m.CreatedAt = t
+}

internal/infrastructure/config/pipelines.go

@@ -52,4 +52,12 @@ var Pipelines = []PipelineDef{
 		IDField:                "professor_id",
 		StateID:                5,
 	},
+	{
+		Name:                   "mensajes",
+		PGQuery:                postgres.MensajeQuery,
+		MongoCollection:        "c_mensajes",
+		MongoDeletedCollection: "deleted_c_mensajes",
+		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/mensaje_query.go

+package postgres
+
+// MensajeQuery arma un documento por matrícula destinataria con los mensajes activos
+// (mensaje_estado = 1) que un director le envió desde Trismegisto. Los mensajes se
+// ordenan por fecha de registro descendente (más reciente primero).
+const MensajeQuery = `
+WITH mensajes_alumno AS (
+    SELECT d.matricula_id,
+           m.mensaje_id,
+           m.mensaje_titulo,
+           m.mensaje_cuerpo,
+           m.mensaje_imagen_url,
+           m.mensaje_fecha_registro,
+           UPPER(pp.persona_nombre || ' ' || pp.persona_apellido_paterno || ' ' || pp.persona_apellido_materno) AS remitente_nombre,
+           s.sede_nombre
+    FROM intranet.mensaje_destinatario d
+             JOIN intranet.mensaje m ON m.mensaje_id = d.mensaje_id AND m.mensaje_estado = 1
+             LEFT JOIN administracion.ad_usuario au ON au.usuario_id = m.usuario_id_remitente
+             LEFT JOIN administracion.ad_empleado ae ON ae.empleado_id = au.empleado_id
+             LEFT JOIN persona.pe_persona pp ON pp.persona_id = ae.persona_id
+             LEFT JOIN administracion.ad_sede s ON s.sede_id = m.sede_id
+)
+SELECT matricula_id,
+       JSONB_AGG(JSONB_BUILD_OBJECT(
+               'mensaje_id', mensaje_id,
+               'titulo', mensaje_titulo,
+               'cuerpo', mensaje_cuerpo,
+               'imagen_url', mensaje_imagen_url,
+               'fecha', to_char(mensaje_fecha_registro, 'YYYY-MM-DD"T"HH24:MI:SS'),
+               'remitente_nombre', remitente_nombre,
+               'sede_nombre', sede_nombre
+           ) ORDER BY mensaje_fecha_registro DESC) AS mensajes
+FROM mensajes_alumno
+GROUP BY matricula_id
+ORDER BY matricula_id
+`
+
+// MensajeByIDQuery es MensajeQuery acotado a una sola matrícula, para la carga parcial
+// que Trismegisto dispara al enviar un mensaje (POST /api/v1/sync/mensajes/upsert).
+const MensajeByIDQuery = `
+WITH mensajes_alumno AS (
+    SELECT d.matricula_id,
+           m.mensaje_id,
+           m.mensaje_titulo,
+           m.mensaje_cuerpo,
+           m.mensaje_imagen_url,
+           m.mensaje_fecha_registro,
+           UPPER(pp.persona_nombre || ' ' || pp.persona_apellido_paterno || ' ' || pp.persona_apellido_materno) AS remitente_nombre,
+           s.sede_nombre
+    FROM intranet.mensaje_destinatario d
+             JOIN intranet.mensaje m ON m.mensaje_id = d.mensaje_id AND m.mensaje_estado = 1
+             LEFT JOIN administracion.ad_usuario au ON au.usuario_id = m.usuario_id_remitente
+             LEFT JOIN administracion.ad_empleado ae ON ae.empleado_id = au.empleado_id
+             LEFT JOIN persona.pe_persona pp ON pp.persona_id = ae.persona_id
+             LEFT JOIN administracion.ad_sede s ON s.sede_id = m.sede_id
+    WHERE d.matricula_id = $1
+)
+SELECT matricula_id,
+       JSONB_AGG(JSONB_BUILD_OBJECT(
+               'mensaje_id', mensaje_id,
+               'titulo', mensaje_titulo,
+               'cuerpo', mensaje_cuerpo,
+               'imagen_url', mensaje_imagen_url,
+               'fecha', to_char(mensaje_fecha_registro, 'YYYY-MM-DD"T"HH24:MI:SS'),
+               'remitente_nombre', remitente_nombre,
+               'sede_nombre', sede_nombre
+           ) ORDER BY mensaje_fecha_registro DESC) AS mensajes
+FROM mensajes_alumno
+GROUP BY matricula_id
+`

internal/infrastructure/database/postgres/mensaje_reader.go

+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+	"intranet-synchronizer/pkg/jsonutil"
+)
+
+type MensajeReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewMensajeReader(pool *Pool, query string) repositories.SourceReader {
+	return &MensajeReader{pool: pool, query: query}
+}
+
+func NewMensajeRowReader(pool *Pool) repositories.SourceRowReader {
+	return &MensajeReader{pool: pool, query: MensajeByIDQuery}
+}
+
+func scanMensajeRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var (
+		matriculaID int
+		mensajesRaw []byte
+	)
+	if err := row.Scan(&matriculaID, &mensajesRaw); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan mensaje row: %w", err)
+	}
+
+	mensajes, err := jsonutil.DecodeList[entities.MensajeItem](mensajesRaw)
+	if err != nil {
+		return entities.SyncRow{}, fmt.Errorf("decode mensajes for matricula %d: %w", matriculaID, err)
+	}
+
+	doc := &entities.Mensaje{
+		MatriculaID: matriculaID,
+		Mensajes:    mensajes,
+	}
+	return entities.SyncRow{ID: matriculaID, Doc: doc}, nil
+}
+
+func (f *MensajeReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query mensajes: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanMensajeRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate mensaje rows: %w", err)
+	}
+	return result, nil
+}
+
+func (f *MensajeReader) FetchByID(ctx context.Context, id int) (entities.SyncRow, bool, error) {
+	rows, err := f.pool.Pool.Query(ctx, MensajeByIDQuery, id)
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("query mensaje %d: %w", id, err)
+	}
+
+	row, err := pgx.CollectExactlyOneRow(rows, scanMensajeRow)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return entities.SyncRow{}, false, nil
+	}
+	if err != nil {
+		return entities.SyncRow{}, false, fmt.Errorf("read mensaje %d: %w", id, err)
+	}
+	return row, true, nil
+}

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 "mensajes":
+		return postgres.NewMensajeReader(pool, query), nil
 	default:
 		return nil, fmt.Errorf("no reader wired for pipeline %q", name)
 	}
@@ -37,6 +39,8 @@ func rowReaderFor(name string, pool *postgres.Pool) repositories.SourceRowReader
 	switch name {
 	case "users":
 		return postgres.NewUserRowReader(pool)
+	case "mensajes":
+		return postgres.NewMensajeRowReader(pool)
 	default:
 		return nil
 	}