Skip to content

I
intranet-synchronizer
Commit 5bfa78f5 by Mauro Paolo Josue Zuñiga Mallqui
Options

refactor: reestructurar a arquitectura limpia 

Reorganiza internal/ en capas: domain (entities, repositories),
usecase (run_sync_pipeline, scheduler) e infrastructure (config,
database/postgres, database/mongo, http). El cableado por pipeline
que vivia inline en main.go pasa a internal/infrastructure/pipeline,
de modo que agregar un pipeline ya no obliga a tocar main.

Agrega health handler y pkg/changelog; jsonutil se mueve a pkg/.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
parent 9f603c1b
Showing with 1044 additions and 725 deletions
README.md
......@@ -6,7 +6,7 @@ Sincroniza estudiantes (con padres y planes de pago) desde PostgreSQL hacia Mong
1. La tabla `sync_state` se crea/inicializa automáticamente al arrancar (fila `id=1` estudiantes) — no se requiere paso de migración manual (`migrations/0001_create_sync_state.sql` se conserva como referencia).
2. Asegúrate de que la colección `students` exista en Mongo con el validador `$jsonSchema` provisto (requeridos: `student_id` int, `enrollment_id` int). La colección `deleted_students` archiva los estudiantes eliminados en el origen — no requiere configuración de esquema.
3. Define las variables de entorno: `PG_DSN`, `MONGO_URI`, `MONGO_DB=intranet` (o tu base de datos), `SYNC_INTERVAL` (por defecto `5m`), `PORT` (por defecto `8080`). Los nombres de función PG / colección Mongo por pipeline ya no son variables de entorno — se definen en `internal/config/pipelines.go`. Agrega un nuevo pipeline añadiendo una entrada ahí, no variables de entorno.
3. Define las variables de entorno: `PG_DSN`, `MONGO_URI`, `MONGO_DB=intranet` (o tu base de datos), `SYNC_INTERVAL` (por defecto `5m`), `PORT` (por defecto `8080`). Los nombres de función PG / colección Mongo por pipeline ya no son variables de entorno — se definen en `internal/infrastructure/config/pipelines.go`. Agrega un nuevo pipeline añadiendo una entrada ahí, no variables de entorno.
4. Horario por pipeline (opcional, `SYNC_<NOMBRE>_...` en mayúsculas, ej. `SYNC_STUDENTS_MODE`, `SYNC_PAYMENT_PLANS_MODE`, `SYNC_PARENTS_MODE`):
- `SYNC_<NOMBRE>_MODE`: `interval` (por defecto) o `daily`.
- `SYNC_<NOMBRE>_INTERVAL`: duración tipo `5m`/`1h` (modo `interval`; si falta, usa `SYNC_INTERVAL`).
......@@ -23,15 +23,26 @@ Todos los endpoints llevan el prefijo `/api/v1`.
- `POST /api/v1/sync/parents/trigger` / `GET /api/v1/sync/parents/status` — pipeline de padres/apoderados (misma forma).
- `POST /api/v1/sync/users/trigger` / `GET /api/v1/sync/users/status` — pipeline de usuarios (misma forma).
## Estructura del proyecto (por colección/módulo)
## Estructura del proyecto (arquitectura limpia)
Cada colección (`students`, `payment_plans`, `parents`, `users`) es un paquete Go propio bajo `internal/<módulo>` (`internal/students`, `internal/paymentplans`, `internal/parents`, `internal/users`):
Mismas capas que `intranet-so-be`:
- `entity.go` — struct del documento Mongo (implementa `SetUpdatedAt` y `SetRowHash`).
- `query.go` — `const Query`, el SQL crudo de origen.
- `reader.go` — `QueryReader`/`NewQueryReader`.
```
cmd/server/main.go raíz de composición
pkg/jsonutil/ helpers genéricos reutilizables
internal/domain/entities/ entidades del dominio + SyncRow/SyncResult
internal/domain/repositories/ puertos (interfaces) que implementa la infraestructura
internal/usecase/ RunSyncPipelineUseCase, Scheduler, Schedule
internal/infrastructure/config/ env vars + Pipelines []PipelineDef
internal/infrastructure/database/postgres/ Pool, SQL por módulo, readers, SyncStateRepository
internal/infrastructure/database/mongo/ cliente + repository/ (upsert y archivado)
internal/infrastructure/http/ handlers/ y router/
internal/infrastructure/pipeline/ builder.go: arma cada pipeline de punta a punta
```
Agregar un módulo nuevo = paquete `internal/<módulo>` + entrada en `internal/config.Pipelines` + case en `cmd/server/main.go` + endpoints en `internal/api`, sin nuevas variables de entorno. Las capas transversales (`sync.Service`, `sync.Scheduler`, `sync.StateStore`, `db.CollectionUpserter`, `db.DeletedMover`) son genéricas y se comparten entre módulos.
Dirección de dependencias: `handlers -> usecase -> domain/repositories`, con la infraestructura implementando esos puertos. El dominio no importa Gin, pgx ni el driver de Mongo.
Agregar un módulo nuevo = entidad en `internal/domain/entities` + `<módulo>_query.go` y `<módulo>_reader.go` en `internal/infrastructure/database/postgres` + entrada en `config.Pipelines` + `case` en `pipeline.readerFor`. Los endpoints `trigger`/`status` se registran solos. Sin nuevas variables de entorno (salvo horario propio opcional).
## Regla de `_id` en Mongo
......
cmd/server/main.go
......@@ -13,14 +13,11 @@ import (
"github.com/joho/godotenv"
"intranet-synchronizer/internal/api"
"intranet-synchronizer/internal/config"
"intranet-synchronizer/internal/db"
"intranet-synchronizer/internal/parents"
"intranet-synchronizer/internal/professors"
"intranet-synchronizer/internal/students"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/users"
"intranet-synchronizer/internal/infrastructure/config"
"intranet-synchronizer/internal/infrastructure/database/mongo"
"intranet-synchronizer/internal/infrastructure/database/postgres"
"intranet-synchronizer/internal/infrastructure/http/router"
"intranet-synchronizer/internal/infrastructure/pipeline"
)
func main() {
......@@ -33,74 +30,37 @@ func main() {
log.Fatalf("config error: %v", err)
}
pgPool, err := db.NewPostgresPool(bootCtx, cfg.PGDSN)
pool, err := postgres.NewPool(bootCtx, cfg.PGDSN)
if err != nil {
log.Fatalf("postgres connect error: %v", err)
}
if err := pgPool.BootstrapSchema(bootCtx); err != nil {
if err := pool.BootstrapSchema(bootCtx); err != nil {
log.Fatalf("schema bootstrap error: %v", err)
}
mongoClient, err := db.NewMongoClient(bootCtx, cfg.MongoURI)
mongoClient, err := mongo.NewClient(bootCtx, cfg.MongoURI)
if err != nil {
log.Fatalf("mongo connect error: %v", err)
}
svcByName := make(map[string]*appsync.Service, len(config.Pipelines))
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
// Por cada pipeline declarado en config.Pipelines, armar su cadena reader/upserter/mover/state/Service/Scheduler.
for _, p := range config.Pipelines {
coll := mongoClient.Database(cfg.MongoDB).Collection(p.MongoCollection)
deletedColl := mongoClient.Database(cfg.MongoDB).Collection(p.MongoDeletedCollection)
var reader appsync.PGReader
switch p.Name {
case "students":
reader = students.NewQueryReader(pgPool, p.PGQuery)
case "parents":
reader = parents.NewQueryReader(pgPool, p.PGQuery)
case "users":
reader = users.NewQueryReader(pgPool, p.PGQuery)
case "professors":
reader = professors.NewQueryReader(pgPool, p.PGQuery)
default:
log.Fatalf("no reader wired for pipeline %q", p.Name)
}
// upserter cumple a la vez MongoUpserter y MongoIDLister.
upserter := db.NewCollectionUpserter(coll, p.IDField)
mover := db.NewDeletedMover(coll, deletedColl, p.IDField)
state := appsync.NewPGStateStore(pgPool, p.StateID)
svc := appsync.NewService(reader, upserter, upserter, mover, state, time.Now)
// Sembrar el último resultado desde el estado persistido antes del primer ciclo.
if lastSynced, err := state.Get(bootCtx); err != nil {
log.Printf("could not load persisted %s sync state at boot, skipping seed: %v", p.Name, err)
} else if !lastSynced.IsZero() {
svc.SeedLastResult(lastSynced)
}
schedule, err := config.ScheduleFor(p.Name, cfg.SyncInterval)
if err != nil {
log.Fatalf("schedule config error for pipeline %s: %v", p.Name, err)
}
scheduler := appsync.NewScheduler(svc, schedule)
go scheduler.Start(ctx)
svcByName[p.Name] = svc
useCases, schedulers, err := pipeline.Build(bootCtx, cfg, pool, mongoClient)
if err != nil {
log.Fatalf("pipeline wiring error: %v", err)
}
for _, s := range schedulers {
go s.Start(ctx)
}
router := api.NewRouter(svcByName["students"], svcByName["parents"], svcByName["users"], svcByName["professors"],
func(pingCtx context.Context) error { return pgPool.Ping(pingCtx) },
func(pingCtx context.Context) error { return mongoClient.Ping(pingCtx, nil) },
)
r := router.New(router.Config{
SyncUseCases: useCases,
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: router}
srv := &http.Server{Addr: ":" + cfg.Port, Handler: r}
go func() {
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("server error: %v", err)
......
internal/api/handlers.go deleted 100644 → 0
package api
import (
"context"
"net/http"
"github.com/gin-gonic/gin"
appsync "intranet-synchronizer/internal/sync"
)
// Handlers guarda las dependencias de los manejadores HTTP.
type Handlers struct {
studentsSvc *appsync.Service
parentsSvc *appsync.Service
usersSvc *appsync.Service
professorsSvc *appsync.Service
pgPing func(context.Context) error
mongoPing func(context.Context) error
}
// Health responde GET /health con el estado de Postgres y Mongo.
func (h *Handlers) Health(c *gin.Context) {
pgErr := h.pgPing(c.Request.Context())
mongoErr := h.mongoPing(c.Request.Context())
status := http.StatusOK
body := gin.H{"postgres": "ok", "mongo": "ok"}
if pgErr != nil {
status = http.StatusServiceUnavailable
body["postgres"] = pgErr.Error()
}
if mongoErr != nil {
status = http.StatusServiceUnavailable
body["mongo"] = mongoErr.Error()
}
c.JSON(status, body)
}
// syncResultJSON arma el cuerpo JSON común de /trigger y /status.
func syncResultJSON(err error, result appsync.SyncResult) gin.H {
resp := gin.H{
"ran_at": result.RanAt,
"rows_synced": result.RowsSynced,
"created": result.Created,
"updated": result.Updated,
"unchanged": result.Unchanged,
"deleted": result.Deleted,
}
if err != nil {
resp["error"] = err.Error()
}
return resp
}
// triggerHandler arma el handler de POST /sync/<pipeline>/trigger: corre una sincronización ahora.
func triggerHandler(svc *appsync.Service) gin.HandlerFunc {
return func(c *gin.Context) {
err := svc.Run(c.Request.Context())
c.JSON(http.StatusOK, syncResultJSON(err, svc.LastResult()))
}
}
// statusHandler arma el handler de GET /sync/<pipeline>/status: devuelve el resultado del último ciclo.
func statusHandler(svc *appsync.Service) gin.HandlerFunc {
return func(c *gin.Context) {
result := svc.LastResult()
resp := syncResultJSON(nil, result)
if result.Err != nil {
resp["last_error"] = result.Err.Error()
}
c.JSON(http.StatusOK, resp)
}
}
internal/api/router.go deleted 100644 → 0
// Package api es la capa web (HTTP), construida con Gin.
package api
import (
"context"
"github.com/gin-gonic/gin"
appsync "intranet-synchronizer/internal/sync"
)
// NewRouter construye el router de Gin y registra las rutas.
func NewRouter(studentsSvc, parentsSvc, usersSvc, professorsSvc *appsync.Service, pgPing, mongoPing func(context.Context) error) *gin.Engine {
h := &Handlers{studentsSvc: studentsSvc, parentsSvc: parentsSvc, usersSvc: usersSvc, professorsSvc: professorsSvc, pgPing: pgPing, mongoPing: mongoPing}
r := gin.Default()
v1 := r.Group("/api/v1")
v1.GET("/health", h.Health)
v1.POST("/sync/students/trigger", triggerHandler(studentsSvc))
v1.GET("/sync/students/status", statusHandler(studentsSvc))
v1.POST("/sync/parents/trigger", triggerHandler(parentsSvc))
v1.GET("/sync/parents/status", statusHandler(parentsSvc))
v1.POST("/sync/users/trigger", triggerHandler(usersSvc))
v1.GET("/sync/users/status", statusHandler(usersSvc))
v1.POST("/sync/professors/trigger", triggerHandler(professorsSvc))
v1.GET("/sync/professors/status", statusHandler(professorsSvc))
return r
}
internal/parents/entity.go → internal/domain/entities/parent.go
// Package parents contiene la entidad, query SQL y reader de la colección "parents".
package parents
package entities
import "time"
// StudentRef es un estudiante asociado a un padre/apoderado, embebido dentro de Record.
// StudentRef es un estudiante asociado a un padre/apoderado, embebido dentro de Parent.
type StudentRef struct {
StudentID int `bson:"student_id" json:"student_id"`
}
// Record es la entidad de la colección "parents".
type Record struct {
// Parent es la entidad de la colección "parents".
type Parent struct {
ParentID int `bson:"parent_id"`
DNI *string `bson:"parent_dni"`
PaternalLastName *string `bson:"parent_paternal_last_name"`
......@@ -23,11 +22,11 @@ type Record struct {
}
// SetUpdatedAt estampa la hora de sincronización.
func (r *Record) SetUpdatedAt(t time.Time) {
func (r *Parent) SetUpdatedAt(t time.Time) {
r.UpdatedAt = t
}
// SetRowHash guarda el hash de contenido usado para saltar upserts sin cambios.
func (r *Record) SetRowHash(h string) {
func (r *Parent) SetRowHash(h string) {
r.RowHash = h
}
internal/professors/entity.go → internal/domain/entities/professor.go
// Package professors contiene la entidad, query SQL y reader de la colección "professors".
package professors
package entities
import "time"
// Record es la entidad de la colección "professors".
type Record struct {
// Professor es la entidad de la colección "professors".
type Professor struct {
ProfessorID int `bson:"professor_id"`
ProfessorPaternalLastName *string `bson:"professor_paternal_last_name"`
ProfessorMaternalLastName *string `bson:"professor_maternal_last_name"`
......@@ -15,11 +14,11 @@ type Record struct {
}
// SetUpdatedAt estampa la hora de sincronización.
func (r *Record) SetUpdatedAt(t time.Time) {
func (r *Professor) SetUpdatedAt(t time.Time) {
r.UpdatedAt = t
}
// SetRowHash guarda el hash de contenido usado para saltar upserts sin cambios.
func (r *Record) SetRowHash(h string) {
func (r *Professor) SetRowHash(h string) {
r.RowHash = h
}
internal/students/entity.go → internal/domain/entities/student.go
// Package students contiene la entidad, query SQL y reader de la colección "students".
package students
// Package entities contiene las entidades de dominio del servicio.
package entities
import "time"
......@@ -14,13 +14,9 @@ type Student struct {
DNI *string `bson:"student_dni"`
Birthday *time.Time `bson:"student_birthday"`
Email *string `bson:"student_email"`
GenreID *int `bson:"genre_id"`
Genre *string `bson:"genre"`
BranchID *int `bson:"branch_id"`
Branch *string `bson:"branch"`
LevelID *int `bson:"level_id"`
Level *string `bson:"level"`
GradeID *int `bson:"grade_id"`
Grade *string `bson:"grade"`
ClassroomAttendanceID *int `bson:"classroom_attendance_id"`
PaymentPlans []PaymentPlanYear `bson:"payment_plans"`
......
internal/domain/entities/sync.go 0 → 100644
package entities
import "time"
// SyncRow es una fila lista para volcar a Mongo: su id de negocio y su documento.
type SyncRow struct {
ID int
Doc interface{}
}
// SyncResult resume cómo salió el último ciclo de un pipeline.
type SyncResult struct {
RanAt time.Time
RowsSynced int
Created int
Updated int
Unchanged int
Deleted int
Err error
}
// UpdatedAtSetter lo implementan las entidades que quieren que el caso de uso les estampe updated_at.
type UpdatedAtSetter interface {
SetUpdatedAt(t time.Time)
}
// RowHashSetter lo implementan las entidades que quieren que el caso de uso les estampe row_hash.
type RowHashSetter interface {
SetRowHash(h string)
}
internal/users/entity.go → internal/domain/entities/user.go
// Package users contiene la entidad, query SQL y reader de la colección "users".
package users
package entities
import "time"
// Record es la entidad de la colección "users".
type Record struct {
// User es la entidad de la colección "users".
type User struct {
UserID int `bson:"user_id"`
Login *string `bson:"user_login"`
Password *string `bson:"user_password"`
......@@ -16,11 +15,11 @@ type Record struct {
}
// SetUpdatedAt estampa la hora de sincronización.
func (r *Record) SetUpdatedAt(t time.Time) {
func (r *User) SetUpdatedAt(t time.Time) {
r.UpdatedAt = t
}
// SetRowHash guarda el hash de contenido usado para saltar upserts sin cambios.
func (r *Record) SetRowHash(h string) {
func (r *User) SetRowHash(h string) {
r.RowHash = h
}
internal/domain/repositories/repositories.go 0 → 100644
package repositories
import "context"
// DocumentFetcher trae los documentos actuales de la colección destino por id de negocio.
// Se usa para registrar en el log de cambios el JSON anterior de las filas que se actualizan.
type DocumentFetcher interface {
FindByIDs(ctx context.Context, ids []int) (map[int]map[string]interface{}, error)
}
internal/domain/repositories/sync_repository.go 0 → 100644
// Package repositories declara los contratos (puertos) que la capa de infraestructura implementa.
package repositories
import (
"context"
"time"
"intranet-synchronizer/internal/domain/entities"
)
// SourceReader lee todas las filas del origen (Postgres) en un ciclo.
type SourceReader interface {
FetchAll(ctx context.Context) ([]entities.SyncRow, error)
}
// DocumentUpserter inserta o actualiza un lote de documentos en una sola operación bulk,
// devolviendo cuántos fueron inserciones nuevas (created) y cuántos actualizaciones (updated).
type DocumentUpserter interface {
BulkUpsert(ctx context.Context, rows []entities.SyncRow) (created, updated int, err error)
}
// DocumentLister lista los ids actuales en la colección destino junto a su row_hash guardado.
// Las claves del mapa detectan borrados; los valores permiten saltar upserts sin cambios.
type DocumentLister interface {
ListIDsAndHashes(ctx context.Context) (map[int]string, error)
}
// DeletedArchiver archiva un registro que ya no está en el origen a la colección de borrados.
type DeletedArchiver interface {
MoveToDeleted(ctx context.Context, id int) error
}
// SyncStateRepository guarda y recupera cuándo fue la última sincronización exitosa.
type SyncStateRepository interface {
Get(ctx context.Context) (time.Time, error)
Set(ctx context.Context, t time.Time) error
}
internal/config/config.go → internal/infrastructure/config/config.go
......@@ -8,7 +8,7 @@ import (
"strings"
"time"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/usecase"
)
// Config guarda los valores de configuración ya leídos y validados.
......@@ -58,7 +58,7 @@ func Load() (Config, error) {
}
// ScheduleFor arma el Schedule de un pipeline a partir de sus variables de entorno opcionales (SYNC_<NOMBRE>_MODE/_INTERVAL/_TIME).
func ScheduleFor(pipelineName string, defaultInterval time.Duration) (appsync.Schedule, error) {
func ScheduleFor(pipelineName string, defaultInterval time.Duration) (usecase.Schedule, error) {
prefix := "SYNC_" + strings.ToUpper(pipelineName)
mode := os.Getenv(prefix + "_MODE")
......@@ -76,7 +76,7 @@ func ScheduleFor(pipelineName string, defaultInterval time.Duration) (appsync.Sc
}
interval = d
}
return appsync.IntervalSchedule{Interval: interval}, nil
return usecase.IntervalSchedule{Interval: interval}, nil
case "daily":
raw := os.Getenv(prefix + "_TIME")
......@@ -87,7 +87,7 @@ func ScheduleFor(pipelineName string, defaultInterval time.Duration) (appsync.Sc
if err != nil {
return nil, fmt.Errorf("invalid %s_TIME %q: %w", prefix, raw, err)
}
return appsync.DailySchedule{Hour: hour, Minute: minute}, nil
return usecase.DailySchedule{Hour: hour, Minute: minute}, nil
default:
return nil, fmt.Errorf("invalid %s_MODE %q: debe ser \"interval\" o \"daily\"", prefix, mode)
......
internal/config/pipelines.go → internal/infrastructure/config/pipelines.go
package config
import (
"intranet-synchronizer/internal/parents"
"intranet-synchronizer/internal/professors"
"intranet-synchronizer/internal/students"
"intranet-synchronizer/internal/users"
)
import "intranet-synchronizer/internal/infrastructure/database/postgres"
// PipelineDef describe una tubería de sincronización: origen Postgres -> destino Mongo.
type PipelineDef struct {
......@@ -21,7 +16,7 @@ type PipelineDef struct {
var Pipelines = []PipelineDef{
{
Name: "students",
PGQuery: students.Query,
PGQuery: postgres.StudentQuery,
MongoCollection: "students",
MongoDeletedCollection: "deleted_students",
IDField: "student_id",
......@@ -29,7 +24,7 @@ var Pipelines = []PipelineDef{
},
{
Name: "parents",
PGQuery: parents.Query,
PGQuery: postgres.ParentQuery,
MongoCollection: "parents",
MongoDeletedCollection: "deleted_parents",
IDField: "parent_id",
......@@ -37,7 +32,7 @@ var Pipelines = []PipelineDef{
},
{
Name: "users",
PGQuery: users.Query,
PGQuery: postgres.UserQuery,
MongoCollection: "users",
MongoDeletedCollection: "deleted_users",
IDField: "user_id",
......@@ -45,7 +40,7 @@ var Pipelines = []PipelineDef{
},
{
Name: "professors",
PGQuery: professors.Query,
PGQuery: postgres.ProfessorQuery,
MongoCollection: "professors",
MongoDeletedCollection: "deleted_professors",
IDField: "professor_id",
......
internal/infrastructure/database/mongo/client.go 0 → 100644
// Package mongo contiene el cliente concreto de MongoDB.
package mongo
import (
"context"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
// NewClient conecta a MongoDB y verifica con un Ping.
func NewClient(ctx context.Context, uri string) (*mongo.Client, error) {
client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
if err != nil {
return nil, err
}
if err := client.Ping(ctx, nil); err != nil {
return nil, err
}
return client, nil
}
internal/infrastructure/database/mongo/repository/deleted_repository.go 0 → 100644
package repository
import (
"context"
"errors"
"fmt"
"time"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"intranet-synchronizer/internal/domain/repositories"
)
// DeletedRepository archiva un registro desaparecido del origen a su colección de borrados, estampando deleted_at.
type DeletedRepository struct {
source *mongo.Collection
deleted *mongo.Collection
idField string
}
func NewDeletedRepository(source, deleted *mongo.Collection, idField string) repositories.DeletedArchiver {
return &DeletedRepository{source: source, deleted: deleted, idField: idField}
}
// MoveToDeleted copia el documento a la colección de borrados y lo elimina de la viva (idempotente).
func (m *DeletedRepository) MoveToDeleted(ctx context.Context, id int) error {
var doc bson.M
filter := bson.M{m.idField: id}
err := m.source.FindOne(ctx, filter).Decode(&doc)
if errors.Is(err, mongo.ErrNoDocuments) {
return nil
}
if err != nil {
return fmt.Errorf("find source doc: %w", err)
}
delete(doc, "_id")
doc["deleted_at"] = time.Now().UTC()
if _, err := m.deleted.UpdateOne(ctx, filter, bson.M{"$set": doc}, options.Update().SetUpsert(true)); err != nil {
return fmt.Errorf("archive doc: %w", err)
}
if _, err := m.source.DeleteOne(ctx, filter); err != nil {
return fmt.Errorf("delete source doc: %w", err)
}
return nil
}
internal/db/mongo.go → internal/infrastructure/database/mongo/repository/document_repository.go
package db
// Package repository implementa, sobre MongoDB, los puertos de escritura del dominio.
package repository
import (
"context"
"errors"
"fmt"
"time"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/domain/entities"
)
// NewMongoClient conecta a MongoDB y verifica con un Ping.
func NewMongoClient(ctx context.Context, uri string) (*mongo.Client, error) {
client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
if err != nil {
return nil, err
}
if err := client.Ping(ctx, nil); err != nil {
return nil, err
}
return client, nil
}
// CollectionUpserter escribe en una colección de Mongo, indexando por idField (no por _id).
type CollectionUpserter struct {
// DocumentRepository escribe en una colección de Mongo, indexando por idField (no por _id).
type DocumentRepository struct {
coll *mongo.Collection
idField string
}
func NewCollectionUpserter(coll *mongo.Collection, idField string) *CollectionUpserter {
return &CollectionUpserter{coll: coll, idField: idField}
func NewDocumentRepository(coll *mongo.Collection, idField string) *DocumentRepository {
return &DocumentRepository{coll: coll, idField: idField}
}
// BulkUpsert manda todas las filas en un único BulkWrite (un round trip) en vez de un UpdateOne
// por fila, evitando la latencia de red repetida de escribir de a una.
func (c *CollectionUpserter) BulkUpsert(ctx context.Context, rows []appsync.SyncRow) (created, updated int, err error) {
func (c *DocumentRepository) BulkUpsert(ctx context.Context, rows []entities.SyncRow) (created, updated int, err error) {
if len(rows) == 0 {
return 0, 0, nil
}
......@@ -61,7 +47,7 @@ func (c *CollectionUpserter) BulkUpsert(ctx context.Context, rows []appsync.Sync
// ListIDsAndHashes devuelve, por cada documento existente, su id de negocio y su row_hash
// guardado (vacío si el documento no tiene hash). Se usa para detectar borrados del origen
// (las claves del mapa) y para saltar upserts de filas sin cambios (comparando el hash).
func (c *CollectionUpserter) ListIDsAndHashes(ctx context.Context) (map[int]string, error) {
func (c *DocumentRepository) ListIDsAndHashes(ctx context.Context) (map[int]string, error) {
cur, err := c.coll.Find(ctx, bson.M{}, options.Find().SetProjection(bson.M{c.idField: 1, "row_hash": 1}))
if err != nil {
return nil, err
......@@ -89,36 +75,34 @@ func (c *CollectionUpserter) ListIDsAndHashes(ctx context.Context) (map[int]stri
return hashes, cur.Err()
}
// DeletedMover archiva un registro desaparecido del origen a su colección de borrados, estampando deleted_at.
type DeletedMover struct {
source *mongo.Collection
deleted *mongo.Collection
idField string
}
func NewDeletedMover(source, deleted *mongo.Collection, idField string) *DeletedMover {
return &DeletedMover{source: source, deleted: deleted, idField: idField}
}
// MoveToDeleted copia el documento a la colección de borrados y lo elimina de la viva (idempotente).
func (m *DeletedMover) MoveToDeleted(ctx context.Context, id int) error {
var doc bson.M
filter := bson.M{m.idField: id}
err := m.source.FindOne(ctx, filter).Decode(&doc)
if errors.Is(err, mongo.ErrNoDocuments) {
return nil
// FindByIDs devuelve los documentos actuales de los ids dados, indexados por id de negocio.
func (c *DocumentRepository) FindByIDs(ctx context.Context, ids []int) (map[int]map[string]interface{}, error) {
docs := make(map[int]map[string]interface{}, len(ids))
if len(ids) == 0 {
return docs, nil
}
cur, err := c.coll.Find(ctx, bson.M{c.idField: bson.M{"$in": ids}})
if err != nil {
return fmt.Errorf("find source doc: %w", err)
return nil, err
}
delete(doc, "_id")
doc["deleted_at"] = time.Now().UTC()
defer cur.Close(ctx)
if _, err := m.deleted.UpdateOne(ctx, filter, bson.M{"$set": doc}, options.Update().SetUpsert(true)); err != nil {
return fmt.Errorf("archive doc: %w", err)
}
if _, err := m.source.DeleteOne(ctx, filter); err != nil {
return fmt.Errorf("delete source doc: %w", err)
for cur.Next(ctx) {
var doc bson.M
if err := cur.Decode(&doc); err != nil {
return nil, err
}
var id int
switch v := doc[c.idField].(type) {
case int32:
id = int(v)
case int64:
id = int(v)
default:
continue
}
docs[id] = doc
}
return nil
return docs, cur.Err()
}
internal/db/postgres.go → internal/infrastructure/database/postgres/client.go
// Package db contiene las implementaciones concretas de acceso a datos (PostgreSQL y MongoDB).
package db
// Package postgres contiene el acceso concreto a PostgreSQL: pool de conexiones,
// readers del origen y persistencia del estado de sincronización.
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
appsync "intranet-synchronizer/internal/sync"
)
// Pool envuelve el pool de pgx para poder colgarle métodos propios.
......@@ -15,8 +14,8 @@ type Pool struct {
*pgxpool.Pool
}
// NewPostgresPool abre el pool y verifica la conexión con un Ping.
func NewPostgresPool(ctx context.Context, dsn string) (*Pool, error) {
// NewPool abre el pool y verifica la conexión con un Ping.
func NewPool(ctx context.Context, dsn string) (*Pool, error) {
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
return nil, err
......@@ -27,17 +26,6 @@ func NewPostgresPool(ctx context.Context, dsn string) (*Pool, error) {
return &Pool{pool}, nil
}
// QueryRow adapta el QueryRow de pgx a la interfaz sync.Row del dominio.
func (p *Pool) QueryRow(ctx context.Context, sql string, args ...interface{}) appsync.Row {
return p.Pool.QueryRow(ctx, sql, args...)
}
// Exec ejecuta una sentencia que no devuelve filas (INSERT/UPDATE/DDL).
func (p *Pool) Exec(ctx context.Context, sql string, args ...interface{}) error {
_, err := p.Pool.Exec(ctx, sql, args...)
return err
}
// bootstrapSyncStateSQL crea la tabla sync_state si no existe y siembra las filas iniciales.
const bootstrapSyncStateSQL = `
CREATE TABLE IF NOT EXISTS sync_state (
......@@ -52,11 +40,8 @@ ON CONFLICT (id) DO NOTHING;
// BootstrapSchema prepara la tabla sync_state al inicio del servicio.
func (p *Pool) BootstrapSchema(ctx context.Context) error {
_, err := p.Pool.Exec(ctx, bootstrapSyncStateSQL)
if err != nil {
if _, err := p.Pool.Exec(ctx, bootstrapSyncStateSQL); err != nil {
return fmt.Errorf("bootstrap sync_state schema: %w", err)
}
return nil
}
// Los readers por módulo (students, parents, users, professors) viven en internal/<módulo>.
internal/parents/query.go → internal/infrastructure/database/postgres/parent_query.go
package parents
package postgres
// Query es el SQL crudo de la tubería de padres/apoderados, con estudiantes a cargo embebidos.
const Query = `
// ParentQuery es el SQL crudo de la tubería de padres/apoderados, con estudiantes a cargo embebidos.
const ParentQuery = `
SELECT pp.persona_id as parent_id,
pp.persona_numero_documento_identidad as parent_dni,
public.to_camel_case(pp.persona_apellido_paterno) as parent_paternal_last_name,
......
internal/parents/reader.go → internal/infrastructure/database/postgres/parent_reader.go
package parents
package postgres
import (
"context"
"fmt"
"intranet-synchronizer/internal/db"
"intranet-synchronizer/internal/jsonutil"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/domain/repositories"
"intranet-synchronizer/pkg/jsonutil"
)
// QueryReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type QueryReader struct {
pool *db.Pool
// ParentReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type ParentReader struct {
pool *Pool
query string
}
// NewQueryReader construye el reader de parents sobre el pool dado.
func NewQueryReader(pool *db.Pool, query string) *QueryReader {
return &QueryReader{pool: pool, query: query}
// NewParentReader construye el reader de parents sobre el pool dado.
func NewParentReader(pool *Pool, query string) repositories.SourceReader {
return &ParentReader{pool: pool, query: query}
}
// FetchAll corre la consulta y lee fila por fila.
func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
func (f *ParentReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
rows, err := f.pool.Pool.Query(ctx, f.query)
if err != nil {
return nil, fmt.Errorf("query parents: %w", err)
}
defer rows.Close()
var result []appsync.SyncRow
var result []entities.SyncRow
for rows.Next() {
var (
parentID int
......@@ -44,12 +44,12 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
return nil, fmt.Errorf("scan parent row: %w", err)
}
studentRefs, err := jsonutil.DecodeList[StudentRef](students)
studentRefs, err := jsonutil.DecodeList[entities.StudentRef](students)
if err != nil {
return nil, fmt.Errorf("decode students for parent %d: %w", parentID, err)
}
doc := &Record{
doc := &entities.Parent{
ParentID: parentID,
DNI: dni,
PaternalLastName: paternalLastName,
......@@ -59,7 +59,7 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
Phone: phone,
Students: studentRefs,
}
result = append(result, appsync.SyncRow{ID: parentID, Doc: doc})
result = append(result, entities.SyncRow{ID: parentID, Doc: doc})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate parent rows: %w", err)
......
internal/infrastructure/database/postgres/payment_plan_query.go 0 → 100644
package postgres
// PaymentPlanQuery es el SQL crudo de la tubería de planes de pago: junta caja.ca_plan_de_pago con matricula.ma_matricula.
const PaymentPlanQuery = `
WITH tb_lista_plan_pago AS (select c.plan_de_pago_id as payment_plan_id,
ma.estudiante_id as student_id,
c.matricula_id as enrollment_id,
COALESCE(c.plan_de_pago_num_orden, 0) as payment_plan_order,
UPPER(c.plan_de_pago_concepto) as payment_plan_concept,
c.plan_de_pago_anio as payment_plan_year,
to_char(c.plan_de_pago_fecha_pago, 'dd-mm-YYYY') as payment_plan_date,
UPPER(COALESCE(c.plan_de_pago_conceptos_adicionales, '') ||
COALESCE(c.plan_de_pago_conceptos_adicionales3, '')) as payment_plan_additional_concepts,
to_char(plan_de_pago_fecha_ven, 'dd-mm-YYYY') as payment_plan_issue_date,
c.plan_de_pago_subtotal as payment_plan_total,
CASE WHEN c.plan_de_pago_deuda THEN 1 ELSE 0 END as payment_plan_debt
FROM caja.ca_plan_de_pago c
INNER JOIN matricula.ma_matricula ma ON c.matricula_id = ma.matricula_id
INNER JOIN academico.ac_apertura ac on ma.apertura_id = ac.apertura_id
WHERE c.plan_de_pago_anulado = false
and ma.matricula_retirado = false
and ma.matricula_anulada = false
and ac.ciclo_id = 28
and ac.periodo_academico_id = 14
ORDER BY to_char(plan_de_pago_fecha_ven, 'YYYY-mm-dd'), plan_de_pago_id)
SELECT *
FROM tb_lista_plan_pago;
`
internal/professors/query.go → internal/infrastructure/database/postgres/professor_query.go
package professors
package postgres
// Query es el SQL crudo de la tubería de profesores.
const Query = `
// ProfessorQuery es el SQL crudo de la tubería de profesores.
const ProfessorQuery = `
SELECT profesor_id as professor_id,
public.to_camel_case(p.persona_apellido_paterno) as professor_paternal_last_name,
public.to_camel_case(p.persona_apellido_materno) as professor_maternal_last_name,
......
internal/professors/reader.go → internal/infrastructure/database/postgres/professor_reader.go
package professors
package postgres
import (
"context"
"fmt"
"intranet-synchronizer/internal/db"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/domain/repositories"
)
// QueryReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type QueryReader struct {
pool *db.Pool
// ProfessorReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type ProfessorReader struct {
pool *Pool
query string
}
// NewQueryReader construye el reader de professors sobre el pool dado.
func NewQueryReader(pool *db.Pool, query string) *QueryReader {
return &QueryReader{pool: pool, query: query}
// NewProfessorReader construye el reader de professors sobre el pool dado.
func NewProfessorReader(pool *Pool, query string) repositories.SourceReader {
return &ProfessorReader{pool: pool, query: query}
}
// FetchAll corre la consulta y lee fila por fila.
func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
func (f *ProfessorReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
rows, err := f.pool.Pool.Query(ctx, f.query)
if err != nil {
return nil, fmt.Errorf("query professors: %w", err)
}
defer rows.Close()
var result []appsync.SyncRow
var result []entities.SyncRow
for rows.Next() {
var (
professorID int
......@@ -40,14 +40,14 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
return nil, fmt.Errorf("scan professor row: %w", err)
}
doc := &Record{
doc := &entities.Professor{
ProfessorID: professorID,
ProfessorPaternalLastName: paternalLastName,
ProfessorMaternalLastName: maternalLastName,
ProfessorName: name,
ProfessorDNI: dni,
}
result = append(result, appsync.SyncRow{ID: professorID, Doc: doc})
result = append(result, entities.SyncRow{ID: professorID, Doc: doc})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate professor rows: %w", err)
......
internal/students/query.go → internal/infrastructure/database/postgres/student_query.go
package students
package postgres
// Query es el SQL crudo de la tubería de estudiantes: junta estudiante + persona + matrícula + apertura (sede/nivel/grado).
// StudentQuery es el SQL crudo de la tubería de estudiantes: junta estudiante + persona + matrícula + apertura (sede/nivel/grado).
// El reader corre además paymentplans.Query por separado y anida sus filas en Student.PaymentPlans por student_id/año
// (ver reader.go) — separado por rendimiento, no se puede traer todo en un solo query sin duplicar filas de estudiante
// por cada plan de pago.
const Query = `
const StudentQuery = `
SELECT e.estudiante_id as student_id,
e.estudiante_codigo as student_code,
e.estudiante_codigo_interno as student_internal_code,
......@@ -27,5 +27,9 @@ FROM matricula.ma_estudiante e
INNER JOIN administracion.ad_sede se on ac.sede_id = se.sede_id
INNER JOIN general.gn_catalogo_siiaa nivel on ac.nivel_id = nivel.catalogo_siiaa_id
INNER JOIN academico.ac_grado grado on ac.grado_id = grado.grado_id
WHERE ma.periodo_academico_id >= 14;
WHERE ac.periodo_academico_id >= 14
and ac.periodo_academico_id = 14
and ac.ciclo_id = 28
and ma.matricula_retirado = false
and ma.matricula_anulada = false;
`
internal/students/reader.go → internal/infrastructure/database/postgres/student_reader.go
package students
package postgres
import (
"context"
......@@ -6,25 +6,24 @@ import (
"sort"
"time"
"intranet-synchronizer/internal/db"
"intranet-synchronizer/internal/paymentplans"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/domain/repositories"
)
// QueryReader ejecuta Query + paymentplans.Query y arma una SyncRow por estudiante,
// StudentReader ejecuta Query + PaymentPlanQuery y arma una SyncRow por estudiante,
// anidando sus planes de pago agrupados por año.
type QueryReader struct {
pool *db.Pool
type StudentReader struct {
pool *Pool
query string
}
// NewQueryReader construye el reader de estudiantes sobre el pool dado.
func NewQueryReader(pool *db.Pool, query string) *QueryReader {
return &QueryReader{pool: pool, query: query}
// NewStudentReader construye el reader de estudiantes sobre el pool dado.
func NewStudentReader(pool *Pool, query string) repositories.SourceReader {
return &StudentReader{pool: pool, query: query}
}
// FetchAll corre la consulta de estudiantes y la de planes de pago, y anida la segunda dentro de la primera.
func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
func (f *StudentReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
students, order, err := f.fetchStudents(ctx)
if err != nil {
return nil, err
......@@ -35,24 +34,24 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
return nil, err
}
result := make([]appsync.SyncRow, 0, len(order))
result := make([]entities.SyncRow, 0, len(order))
for _, studentID := range order {
doc := students[studentID]
doc.PaymentPlans = buildPaymentPlanYears(plansByStudent[studentID])
result = append(result, appsync.SyncRow{ID: studentID, Doc: doc})
result = append(result, entities.SyncRow{ID: studentID, Doc: doc})
}
return result, nil
}
// fetchStudents lee la fila base de cada estudiante (sin planes de pago).
func (f *QueryReader) fetchStudents(ctx context.Context) (map[int]*Student, []int, error) {
func (f *StudentReader) fetchStudents(ctx context.Context) (map[int]*entities.Student, []int, error) {
rows, err := f.pool.Pool.Query(ctx, f.query)
if err != nil {
return nil, nil, fmt.Errorf("query students: %w", err)
}
defer rows.Close()
students := make(map[int]*Student)
students := make(map[int]*entities.Student)
var order []int
for rows.Next() {
var (
......@@ -65,23 +64,19 @@ func (f *QueryReader) fetchStudents(ctx context.Context) (map[int]*Student, []in
dni *string
birthday *time.Time
email *string
genreID *int
genre *string
branchID *int
branch *string
levelID *int
level *string
gradeID *int
grade *string
classroomAttendanceID *int
)
if err := rows.Scan(&studentID, &studentCode, &studentInternalCode, &paternalLastName, &maternalLastName,
&name, &dni, &birthday, &email, &genreID, &genre, &branchID, &branch, &levelID, &level,
&gradeID, &grade, &classroomAttendanceID); err != nil {
&name, &dni, &birthday, &email, &genre, &branch, &level,
&grade, &classroomAttendanceID); err != nil {
return nil, nil, fmt.Errorf("scan student row: %w", err)
}
students[studentID] = &Student{
students[studentID] = &entities.Student{
StudentID: studentID,
StudentCode: studentCode,
StudentInternalCode: studentInternalCode,
......@@ -91,13 +86,9 @@ func (f *QueryReader) fetchStudents(ctx context.Context) (map[int]*Student, []in
DNI: dni,
Birthday: birthday,
Email: email,
GenreID: genreID,
Genre: genre,
BranchID: branchID,
Branch: branch,
LevelID: levelID,
Level: level,
GradeID: gradeID,
Grade: grade,
ClassroomAttendanceID: classroomAttendanceID,
}
......@@ -109,15 +100,15 @@ func (f *QueryReader) fetchStudents(ctx context.Context) (map[int]*Student, []in
return students, order, nil
}
// paymentRow es una fila de paymentplans.Query antes de agruparse por año.
// paymentRow es una fila de PaymentPlanQuery antes de agruparse por año.
type paymentRow struct {
Payment
entities.Payment
year int
}
// fetchPaymentPlans lee todos los planes de pago y los agrupa por student_id.
func (f *QueryReader) fetchPaymentPlans(ctx context.Context) (map[int][]paymentRow, error) {
rows, err := f.pool.Pool.Query(ctx, paymentplans.Query)
func (f *StudentReader) fetchPaymentPlans(ctx context.Context) (map[int][]paymentRow, error) {
rows, err := f.pool.Pool.Query(ctx, PaymentPlanQuery)
if err != nil {
return nil, fmt.Errorf("query payment_plans: %w", err)
}
......@@ -144,7 +135,7 @@ func (f *QueryReader) fetchPaymentPlans(ctx context.Context) (map[int][]paymentR
}
plansByStudent[studentID] = append(plansByStudent[studentID], paymentRow{
Payment: Payment{
Payment: entities.Payment{
PaymentPlanID: paymentPlanID,
EnrollmentID: enrollmentID,
Order: order,
......@@ -165,8 +156,8 @@ func (f *QueryReader) fetchPaymentPlans(ctx context.Context) (map[int][]paymentR
}
// buildPaymentPlanYears agrupa los pagos de un estudiante por año, orden descendente.
func buildPaymentPlanYears(payments []paymentRow) []PaymentPlanYear {
byYear := make(map[int][]Payment)
func buildPaymentPlanYears(payments []paymentRow) []entities.PaymentPlanYear {
byYear := make(map[int][]entities.Payment)
for _, p := range payments {
byYear[p.year] = append(byYear[p.year], p.Payment)
}
......@@ -177,9 +168,9 @@ func buildPaymentPlanYears(payments []paymentRow) []PaymentPlanYear {
}
sort.Sort(sort.Reverse(sort.IntSlice(years)))
result := make([]PaymentPlanYear, 0, len(years))
result := make([]entities.PaymentPlanYear, 0, len(years))
for _, y := range years {
result = append(result, PaymentPlanYear{Year: y, Payments: byYear[y]})
result = append(result, entities.PaymentPlanYear{Year: y, Payments: byYear[y]})
}
return result
}
internal/infrastructure/database/postgres/sync_state_repository.go 0 → 100644
package postgres
import (
"context"
"time"
"intranet-synchronizer/internal/domain/repositories"
)
// SyncStateRepository persiste el timestamp de última sincronización en la tabla sync_state,
// una fila por pipeline (stateID).
type SyncStateRepository struct {
pool *Pool
stateID int
}
// NewSyncStateRepository construye el repositorio de estado de un pipeline.
func NewSyncStateRepository(pool *Pool, stateID int) repositories.SyncStateRepository {
return &SyncStateRepository{pool: pool, stateID: stateID}
}
const selectLastSyncedSQL = `SELECT last_synced_at FROM sync_state WHERE id = $1`
// Get lee el último timestamp sincronizado para este pipeline.
func (s *SyncStateRepository) Get(ctx context.Context) (time.Time, error) {
var t time.Time
if err := s.pool.Pool.QueryRow(ctx, selectLastSyncedSQL, s.stateID).Scan(&t); err != nil {
return time.Time{}, err
}
return t, nil
}
// upsertLastSyncedSQL inserta la fila si no existe o actualiza last_synced_at si ya existe.
const upsertLastSyncedSQL = `
INSERT INTO sync_state (id, last_synced_at) VALUES ($1, $2)
ON CONFLICT (id) DO UPDATE SET last_synced_at = EXCLUDED.last_synced_at`
// Set guarda un nuevo timestamp de última sincronización exitosa.
func (s *SyncStateRepository) Set(ctx context.Context, t time.Time) error {
_, err := s.pool.Pool.Exec(ctx, upsertLastSyncedSQL, s.stateID, t)
return err
}
internal/users/query.go → internal/infrastructure/database/postgres/user_query.go
package users
package postgres
// Query es el SQL crudo de la tubería de usuarios.
const Query = `
// UserQuery es el SQL crudo de la tubería de usuarios.
const UserQuery = `
SELECT user_id, user_login, user_password, user_creation_date, parent_id, user_status
FROM intranet.users;
`
internal/users/reader.go → internal/infrastructure/database/postgres/user_reader.go
package users
package postgres
import (
"context"
"fmt"
"time"
"intranet-synchronizer/internal/db"
appsync "intranet-synchronizer/internal/sync"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/domain/repositories"
)
// QueryReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type QueryReader struct {
pool *db.Pool
// UserReader ejecuta Query directamente y arma una SyncRow por fila del resultado.
type UserReader struct {
pool *Pool
query string
}
// NewQueryReader construye el reader de users sobre el pool dado.
func NewQueryReader(pool *db.Pool, query string) *QueryReader {
return &QueryReader{pool: pool, query: query}
// NewUserReader construye el reader de users sobre el pool dado.
func NewUserReader(pool *Pool, query string) repositories.SourceReader {
return &UserReader{pool: pool, query: query}
}
// FetchAll corre la consulta y lee fila por fila.
func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
func (f *UserReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
rows, err := f.pool.Pool.Query(ctx, f.query)
if err != nil {
return nil, fmt.Errorf("query users: %w", err)
}
defer rows.Close()
var result []appsync.SyncRow
var result []entities.SyncRow
for rows.Next() {
var (
userID int
......@@ -42,7 +42,7 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
return nil, fmt.Errorf("scan user row: %w", err)
}
doc := &Record{
doc := &entities.User{
UserID: userID,
Login: login,
Password: password,
......@@ -50,7 +50,7 @@ func (f *QueryReader) FetchAll(ctx context.Context) ([]appsync.SyncRow, error) {
ParentID: parentID,
Status: status,
}
result = append(result, appsync.SyncRow{ID: userID, Doc: doc})
result = append(result, entities.SyncRow{ID: userID, Doc: doc})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate user rows: %w", err)
......
internal/infrastructure/http/handlers/health_handler.go 0 → 100644
// Package handlers contiene los manejadores HTTP (capa web, Gin).
package handlers
import (
"context"
"net/http"
"github.com/gin-gonic/gin"
)
// HealthHandler responde el chequeo de salud del servicio.
type HealthHandler struct {
pgPing func(context.Context) error
mongoPing func(context.Context) error
}
// NewHealthHandler construye el handler de salud con los pings de cada motor.
func NewHealthHandler(pgPing, mongoPing func(context.Context) error) *HealthHandler {
return &HealthHandler{pgPing: pgPing, mongoPing: mongoPing}
}
// Health responde GET /health con el estado de Postgres y Mongo.
func (h *HealthHandler) Health(c *gin.Context) {
pgErr := h.pgPing(c.Request.Context())
mongoErr := h.mongoPing(c.Request.Context())
status := http.StatusOK
body := gin.H{"postgres": "ok", "mongo": "ok"}
if pgErr != nil {
status = http.StatusServiceUnavailable
body["postgres"] = pgErr.Error()
}
if mongoErr != nil {
status = http.StatusServiceUnavailable
body["mongo"] = mongoErr.Error()
}
c.JSON(status, body)
}
internal/infrastructure/http/handlers/sync_handler.go 0 → 100644
package handlers
import (
"net/http"
"github.com/gin-gonic/gin"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/usecase"
)
// SyncHandler expone los endpoints trigger/status de UN pipeline de sincronización.
type SyncHandler struct {
runSync usecase.RunSyncPipelineUseCase
}
// NewSyncHandler construye el handler de un pipeline sobre su caso de uso.
func NewSyncHandler(runSync usecase.RunSyncPipelineUseCase) *SyncHandler {
return &SyncHandler{runSync: runSync}
}
// syncResultJSON arma el cuerpo JSON común de /trigger y /status.
func syncResultJSON(err error, result entities.SyncResult) gin.H {
resp := gin.H{
"ran_at": result.RanAt,
"rows_synced": result.RowsSynced,
"created": result.Created,
"updated": result.Updated,
"unchanged": result.Unchanged,
"deleted": result.Deleted,
}
if err != nil {
resp["error"] = err.Error()
}
return resp
}
// Trigger atiende POST /sync/<pipeline>/trigger: corre una sincronización ahora.
func (h *SyncHandler) Trigger(c *gin.Context) {
err := h.runSync.Run(c.Request.Context())
c.JSON(http.StatusOK, syncResultJSON(err, h.runSync.LastResult()))
}
// Status atiende GET /sync/<pipeline>/status: devuelve el resultado del último ciclo.
func (h *SyncHandler) Status(c *gin.Context) {
result := h.runSync.LastResult()
resp := syncResultJSON(nil, result)
if result.Err != nil {
resp["last_error"] = result.Err.Error()
}
c.JSON(http.StatusOK, resp)
}
internal/infrastructure/http/router/router.go 0 → 100644
// Package router arma el router HTTP y conecta handlers con casos de uso.
package router
import (
"context"
"github.com/gin-gonic/gin"
"intranet-synchronizer/internal/infrastructure/http/handlers"
"intranet-synchronizer/internal/usecase"
)
// Config son las dependencias ya construidas que el router necesita.
type Config struct {
// SyncUseCases mapea el nombre del pipeline a su caso de uso (ej. "students").
SyncUseCases map[string]usecase.RunSyncPipelineUseCase
PGPing func(context.Context) error
MongoPing func(context.Context) error
}
// New construye el router de Gin y registra las rutas de /api/v1.
func New(cfg Config) *gin.Engine {
r := gin.Default()
health := handlers.NewHealthHandler(cfg.PGPing, cfg.MongoPing)
v1 := r.Group("/api/v1")
v1.GET("/health", health.Health)
// Un par trigger/status por pipeline declarado.
for name, uc := range cfg.SyncUseCases {
h := handlers.NewSyncHandler(uc)
v1.POST("/sync/"+name+"/trigger", h.Trigger)
v1.GET("/sync/"+name+"/status", h.Status)
}
return r
}
internal/infrastructure/pipeline/builder.go 0 → 100644
// Package pipeline arma, por cada pipeline declarado en config.Pipelines, la cadena
// completa reader -> upserter/lister -> archiver -> estado -> caso de uso -> scheduler.
package pipeline
import (
"context"
"fmt"
"time"
mongodriver "go.mongodb.org/mongo-driver/mongo"
"intranet-synchronizer/internal/domain/repositories"
"intranet-synchronizer/internal/infrastructure/config"
"intranet-synchronizer/internal/infrastructure/database/mongo/repository"
"intranet-synchronizer/internal/infrastructure/database/postgres"
"intranet-synchronizer/internal/usecase"
"intranet-synchronizer/pkg/changelog"
)
// readerFor elige el reader de Postgres que corresponde al pipeline.
func readerFor(name string, pool *postgres.Pool, query string) (repositories.SourceReader, error) {
switch name {
case "students":
return postgres.NewStudentReader(pool, query), nil
case "parents":
return postgres.NewParentReader(pool, query), nil
case "users":
return postgres.NewUserReader(pool, query), nil
case "professors":
return postgres.NewProfessorReader(pool, query), nil
default:
return nil, fmt.Errorf("no reader wired for pipeline %q", name)
}
}
// Build arma los casos de uso y schedulers de todos los pipelines declarados.
// Devuelve los casos de uso por nombre (para el router) y los schedulers listos para arrancar.
func Build(
ctx context.Context,
cfg config.Config,
pool *postgres.Pool,
mongoClient *mongodriver.Client,
) (map[string]usecase.RunSyncPipelineUseCase, []*usecase.Scheduler, error) {
useCases := make(map[string]usecase.RunSyncPipelineUseCase, len(config.Pipelines))
schedulers := make([]*usecase.Scheduler, 0, len(config.Pipelines))
for _, p := range config.Pipelines {
db := mongoClient.Database(cfg.MongoDB)
reader, err := readerFor(p.Name, pool, p.PGQuery)
if err != nil {
return nil, nil, err
}
// docRepo cumple a la vez DocumentUpserter y DocumentLister.
docRepo := repository.NewDocumentRepository(db.Collection(p.MongoCollection), p.IDField)
archiver := repository.NewDeletedRepository(db.Collection(p.MongoCollection), db.Collection(p.MongoDeletedCollection), p.IDField)
state := postgres.NewSyncStateRepository(pool, p.StateID)
// changeLog deja en /opt/logs/sync_<pipeline>.log el JSON de cada alta y el par
// anterior/nuevo de cada actualización.
changeLog, err := changelog.New(changelog.DefaultDir, p.Name)
if err != nil {
return nil, nil, fmt.Errorf("change log for pipeline %s: %w", p.Name, err)
}
uc := usecase.NewRunSyncPipelineUseCase(p.Name, reader, docRepo, docRepo, archiver, state, docRepo, changeLog, time.Now)
// Sembrar el último resultado desde el estado persistido antes del primer ciclo.
if lastSynced, err := state.Get(ctx); err != nil {
return nil, nil, fmt.Errorf("load persisted %s sync state: %w", p.Name, err)
} else if !lastSynced.IsZero() {
uc.SeedLastResult(lastSynced)
}
schedule, err := config.ScheduleFor(p.Name, cfg.SyncInterval)
if err != nil {
return nil, nil, fmt.Errorf("schedule config error for pipeline %s: %w", p.Name, err)
}
useCases[p.Name] = uc
schedulers = append(schedulers, usecase.NewScheduler(uc, schedule))
}
return useCases, schedulers, nil
}
internal/paymentplans/query.go deleted 100644 → 0
package paymentplans
// Query es el SQL crudo de la tubería de planes de pago: junta caja.ca_plan_de_pago con matricula.ma_matricula.
const Query = `
WITH tb_lista_plan_pago AS (select c.plan_de_pago_id as payment_plan_id,
m.estudiante_id as student_id,
c.matricula_id as enrollment_id,
COALESCE(c.plan_de_pago_num_orden,0) as payment_plan_order,
UPPER(c.plan_de_pago_concepto) as payment_plan_concept,
c.plan_de_pago_anio as payment_plan_year,
to_char(c.plan_de_pago_fecha_pago, 'dd-mm-YYYY') as payment_plan_date,
UPPER(COALESCE(c.plan_de_pago_conceptos_adicionales, '') ||
COALESCE(c.plan_de_pago_conceptos_adicionales3, '')) as payment_plan_additional_concepts,
to_char(plan_de_pago_fecha_ven, 'dd-mm-YYYY') as payment_plan_issue_date,
c.plan_de_pago_subtotal as payment_plan_total,
CASE WHEN c.plan_de_pago_deuda THEN 1 ELSE 0 END as payment_plan_debt
FROM caja.ca_plan_de_pago c
INNER JOIN matricula.ma_matricula m ON c.matricula_id = m.matricula_id
WHERE plan_de_pago_anulado = false
and m.periodo_academico_id >= 14
ORDER BY to_char(plan_de_pago_fecha_ven, 'YYYY-mm-dd'), plan_de_pago_id)
SELECT *
FROM tb_lista_plan_pago;
`
internal/sync/service.go deleted 100644 → 0
package sync
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sync"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
)
// defaultConcurrency: cuántas filas se procesan a la vez en paralelo.
const defaultConcurrency = 10
// SyncRow es una fila lista para volcar a Mongo: su id y su documento.
type SyncRow struct {
ID int
Doc interface{}
}
// updatedAtSetter lo implementan las entidades tipadas que quieren que el Service les estampe updated_at.
type updatedAtSetter interface {
SetUpdatedAt(t time.Time)
}
// rowHashSetter lo implementan las entidades tipadas que quieren que el Service les estampe row_hash.
type rowHashSetter interface {
SetRowHash(h string)
}
// stampUpdatedAt estampa updated_at en el documento, sea map genérico o entidad tipada.
func stampUpdatedAt(doc interface{}, t time.Time) {
switch v := doc.(type) {
case map[string]interface{}:
v["updated_at"] = t
case updatedAtSetter:
v.SetUpdatedAt(t)
}
}
// stampRowHash guarda el hash de contenido en el documento, sea map genérico o entidad tipada.
func stampRowHash(doc interface{}, h string) {
switch v := doc.(type) {
case map[string]interface{}:
v["row_hash"] = h
case rowHashSetter:
v.SetRowHash(h)
}
}
// computeRowHash calcula un hash de contenido del documento, ANTES de estampar updated_at/row_hash,
// para poder comparar contra el hash guardado en Mongo y saltar el upsert si la fila no cambió.
func computeRowHash(doc interface{}) (string, error) {
b, err := json.Marshal(doc)
if err != nil {
return "", fmt.Errorf("marshal row for hash: %w", err)
}
sum := sha256.Sum256(b)
return hex.EncodeToString(sum[:]), nil
}
// PGReader lee todas las filas del origen (Postgres) en un ciclo.
type PGReader interface {
FetchAll(ctx context.Context) ([]SyncRow, error)
}
// MongoUpserter inserta o actualiza un lote de documentos en una sola operación bulk,
// devolviendo cuántos fueron inserciones nuevas (created) y cuántos actualizaciones (updated).
type MongoUpserter interface {
BulkUpsert(ctx context.Context, rows []SyncRow) (created, updated int, err error)
}
// MongoIDLister lista los ids actuales en la colección Mongo junto a su row_hash guardado.
// Las claves del mapa detectan borrados; los valores permiten saltar upserts sin cambios.
type MongoIDLister interface {
ListIDsAndHashes(ctx context.Context) (map[int]string, error)
}
// DeletedMover archiva un registro que ya no está en Postgres a la colección de borrados.
type DeletedMover interface {
MoveToDeleted(ctx context.Context, id int) error
}
// SyncResult resume cómo salió el último ciclo.
type SyncResult struct {
RanAt time.Time
RowsSynced int
Created int
Updated int
Unchanged int
Deleted int
Err error
}
// Service orquesta un ciclo completo de sincronización de UN pipeline.
type Service struct {
reader PGReader
mongo MongoUpserter
lister MongoIDLister
mover DeletedMover
state StateStore
now func() time.Time
concurrency int
// mu protege lastResult entre goroutines concurrentes.
mu sync.Mutex
lastResult SyncResult
}
// NewService arma un Service inyectándole todas sus dependencias.
func NewService(reader PGReader, mongo MongoUpserter, lister MongoIDLister, mover DeletedMover, state StateStore, now func() time.Time) *Service {
return &Service{
reader: reader,
mongo: mongo,
lister: lister,
mover: mover,
state: state,
now: now,
concurrency: defaultConcurrency,
}
}
// Run ejecuta un ciclo completo: lee Postgres, upsertea en Mongo, archiva borrados,
// avanza el estado persistido solo si todo salió bien.
func (s *Service) Run(ctx context.Context) error {
s.mu.Lock()
defer s.mu.Unlock()
runAt := s.now().UTC()
result := SyncResult{RanAt: runAt}
rows, err := s.reader.FetchAll(ctx)
if err != nil {
result.Err = fmt.Errorf("fetch all: %w", err)
s.lastResult = result
return result.Err
}
pgIDs := make(map[int]struct{}, len(rows))
for _, row := range rows {
pgIDs[row.ID] = struct{}{}
}
// existingHashes trae, en una sola pasada, el row_hash ya guardado por id (para saltar
// upserts de filas sin cambios) y sirve además como base para detectar borrados abajo.
var existingHashes map[int]string
if s.lister != nil {
var err error
existingHashes, err = s.lister.ListIDsAndHashes(ctx)
if err != nil {
result.Err = fmt.Errorf("list existing hashes: %w", err)
s.lastResult = result
return result.Err
}
}
// Fase 1 (CPU, concurrente): calcular el hash de cada fila y descartar las que no cambiaron
// frente a lo guardado en Mongo. No hay I/O acá, solo sirve para no mandar filas de más al bulk.
changed := make([]SyncRow, len(rows))
var changedCount, unchanged int64
g, _ := errgroup.WithContext(ctx)
g.SetLimit(s.concurrency)
for _, row := range rows {
row := row
g.Go(func() error {
hash, err := computeRowHash(row.Doc)
if err != nil {
return fmt.Errorf("hash row %d: %w", row.ID, err)
}
if existingHash, ok := existingHashes[row.ID]; ok && existingHash == hash {
atomic.AddInt64(&unchanged, 1)
return nil
}
stampUpdatedAt(row.Doc, runAt)
stampRowHash(row.Doc, hash)
idx := atomic.AddInt64(&changedCount, 1) - 1
changed[idx] = row
return nil
})
}
if err := g.Wait(); err != nil {
result.Err = err
s.lastResult = result
return result.Err
}
changed = changed[:changedCount]
// Fase 2 (I/O, un solo round trip): escribir todas las filas que cambiaron en un único BulkWrite.
created, updated, err := s.mongo.BulkUpsert(ctx, changed)
if err != nil {
result.Err = fmt.Errorf("bulk upsert: %w", err)
s.lastResult = result
return result.Err
}
result.Created = created
result.Updated = updated
result.Unchanged = int(unchanged)
result.RowsSynced = len(rows)
deleted, err := s.reconcileDeleted(ctx, pgIDs, existingHashes)
if err != nil {
result.Err = fmt.Errorf("reconcile deleted: %w", err)
s.lastResult = result
return result.Err
}
result.Deleted = deleted
if err := s.state.Set(ctx, runAt); err != nil {
result.Err = fmt.Errorf("advance sync state: %w", err)
s.lastResult = result
return result.Err
}
s.lastResult = result
return nil
}
// reconcileDeleted encuentra ids en Mongo (existingHashes) que ya no están en Postgres y los archiva.
func (s *Service) reconcileDeleted(ctx context.Context, pgIDs map[int]struct{}, existingHashes map[int]string) (int, error) {
if s.lister == nil || s.mover == nil {
return 0, nil
}
var toDelete []int
for id := range existingHashes {
if _, ok := pgIDs[id]; !ok {
toDelete = append(toDelete, id)
}
}
if len(toDelete) == 0 {
return 0, nil
}
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(s.concurrency)
for _, id := range toDelete {
id := id
g.Go(func() error {
if err := s.mover.MoveToDeleted(gctx, id); err != nil {
return fmt.Errorf("move student %d to deleted: %w", id, err)
}
return nil
})
}
if err := g.Wait(); err != nil {
return 0, err
}
return len(toDelete), nil
}
// SeedLastResult inicializa lastResult con un timestamp persistido tras arrancar el proceso.
func (s *Service) SeedLastResult(t time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastResult = SyncResult{RanAt: t}
}
// LastResult devuelve una copia del último resultado, protegida por el mutex.
func (s *Service) LastResult() SyncResult {
s.mu.Lock()
defer s.mu.Unlock()
return s.lastResult
}
internal/sync/state.go deleted 100644 → 0
// Package sync es el dominio del servicio: lógica de sincronización sin depender de librerías externas de base de datos ni del framework web.
package sync
import (
"context"
"time"
)
// Row abstrae el resultado de una consulta de una sola fila (lo satisface pgx.Row).
type Row interface {
Scan(dest ...interface{}) error
}
// PGExecutor abstrae el acceso mínimo a Postgres que necesita el state store.
type PGExecutor interface {
QueryRow(ctx context.Context, sql string, args ...interface{}) Row
Exec(ctx context.Context, sql string, args ...interface{}) error
}
// StateStore guarda y recupera cuándo fue la última sincronización exitosa.
type StateStore interface {
Get(ctx context.Context) (time.Time, error)
Set(ctx context.Context, t time.Time) error
}
// PGStateStore persiste el timestamp de última sincronización en la tabla sync_state, una fila por pipeline (stateID).
type PGStateStore struct {
db PGExecutor
stateID int
}
func NewPGStateStore(db PGExecutor, stateID int) *PGStateStore {
return &PGStateStore{db: db, stateID: stateID}
}
const selectLastSyncedSQL = `SELECT last_synced_at FROM sync_state WHERE id = $1`
// Get lee el último timestamp sincronizado para este pipeline.
func (s *PGStateStore) Get(ctx context.Context) (time.Time, error) {
var t time.Time
row := s.db.QueryRow(ctx, selectLastSyncedSQL, s.stateID)
if err := row.Scan(&t); err != nil {
return time.Time{}, err
}
return t, nil
}
// upsertLastSyncedSQL inserta la fila si no existe o actualiza last_synced_at si ya existe.
const upsertLastSyncedSQL = `
INSERT INTO sync_state (id, last_synced_at) VALUES ($1, $2)
ON CONFLICT (id) DO UPDATE SET last_synced_at = EXCLUDED.last_synced_at`
// Set guarda un nuevo timestamp de última sincronización exitosa.
func (s *PGStateStore) Set(ctx context.Context, t time.Time) error {
return s.db.Exec(ctx, upsertLastSyncedSQL, s.stateID, t)
}
internal/usecase/run_sync_pipeline.go 0 → 100644
// Package usecase contiene la lógica de aplicación: orquesta los puertos del dominio
// sin depender de librerías de base de datos ni del framework web.
package usecase
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"sort"
"sync"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
"intranet-synchronizer/internal/domain/entities"
"intranet-synchronizer/internal/domain/repositories"
)
// defaultConcurrency: cuántas filas se procesan a la vez en paralelo.
const defaultConcurrency = 10
// stampUpdatedAt estampa updated_at en el documento, sea map genérico o entidad tipada.
func stampUpdatedAt(doc interface{}, t time.Time) {
switch v := doc.(type) {
case map[string]interface{}:
v["updated_at"] = t
case entities.UpdatedAtSetter:
v.SetUpdatedAt(t)
}
}
// stampRowHash guarda el hash de contenido en el documento, sea map genérico o entidad tipada.
func stampRowHash(doc interface{}, h string) {
switch v := doc.(type) {
case map[string]interface{}:
v["row_hash"] = h
case entities.RowHashSetter:
v.SetRowHash(h)
}
}
// computeRowHash calcula un hash de contenido del documento, ANTES de estampar updated_at/row_hash,
// para poder comparar contra el hash guardado en Mongo y saltar el upsert si la fila no cambió.
func computeRowHash(doc interface{}) (string, error) {
b, err := json.Marshal(doc)
if err != nil {
return "", fmt.Errorf("marshal row for hash: %w", err)
}
sum := sha256.Sum256(b)
return hex.EncodeToString(sum[:]), nil
}
// RunSyncPipelineUseCase ejecuta un ciclo completo de sincronización de UN pipeline
// y expone el resultado del último ciclo.
type RunSyncPipelineUseCase interface {
Run(ctx context.Context) error
SeedLastResult(t time.Time)
LastResult() entities.SyncResult
}
// ChangeLogger registra en un archivo el detalle de cada documento insertado o actualizado.
type ChangeLogger interface {
Insert(id int, doc interface{}) error
Update(id int, before, after interface{}) error
}
// runSyncPipeline es la implementación de RunSyncPipelineUseCase.
type runSyncPipeline struct {
name string
fetcher repositories.DocumentFetcher
changeLog ChangeLogger
reader repositories.SourceReader
upserter repositories.DocumentUpserter
lister repositories.DocumentLister
archiver repositories.DeletedArchiver
state repositories.SyncStateRepository
now func() time.Time
concurrency int
// mu protege lastResult entre goroutines concurrentes.
mu sync.Mutex
lastResult entities.SyncResult
}
// NewRunSyncPipelineUseCase arma el caso de uso inyectándole todos sus puertos.
func NewRunSyncPipelineUseCase(
name string,
reader repositories.SourceReader,
upserter repositories.DocumentUpserter,
lister repositories.DocumentLister,
archiver repositories.DeletedArchiver,
state repositories.SyncStateRepository,
fetcher repositories.DocumentFetcher,
changeLog ChangeLogger,
now func() time.Time,
) RunSyncPipelineUseCase {
return &runSyncPipeline{
name: name,
fetcher: fetcher,
changeLog: changeLog,
reader: reader,
upserter: upserter,
lister: lister,
archiver: archiver,
state: state,
now: now,
concurrency: defaultConcurrency,
}
}
// Run ejecuta un ciclo completo: lee Postgres, upsertea en Mongo, archiva borrados,
// avanza el estado persistido solo si todo salió bien.
func (s *runSyncPipeline) Run(ctx context.Context) error {
s.mu.Lock()
defer s.mu.Unlock()
runAt := s.now().UTC()
result := entities.SyncResult{RanAt: runAt}
rows, err := s.reader.FetchAll(ctx)
if err != nil {
result.Err = fmt.Errorf("fetch all: %w", err)
s.lastResult = result
return result.Err
}
pgIDs := make(map[int]struct{}, len(rows))
for _, row := range rows {
pgIDs[row.ID] = struct{}{}
}
// existingHashes trae, en una sola pasada, el row_hash ya guardado por id (para saltar
// upserts de filas sin cambios) y sirve además como base para detectar borrados abajo.
var existingHashes map[int]string
if s.lister != nil {
var err error
existingHashes, err = s.lister.ListIDsAndHashes(ctx)
if err != nil {
result.Err = fmt.Errorf("list existing hashes: %w", err)
s.lastResult = result
return result.Err
}
}
// Fase 1 (CPU, concurrente): calcular el hash de cada fila y descartar las que no cambiaron
// frente a lo guardado en Mongo. No hay I/O acá, solo sirve para no mandar filas de más al bulk.
changed := make([]entities.SyncRow, len(rows))
var changedCount, unchanged int64
// createdIDs/updatedIDs se acumulan para poder listarlos en consola al cerrar el ciclo.
var (
idsMu sync.Mutex
createdIDs []int
updatedIDs []int
)
g, _ := errgroup.WithContext(ctx)
g.SetLimit(s.concurrency)
for _, row := range rows {
row := row
g.Go(func() error {
hash, err := computeRowHash(row.Doc)
if err != nil {
return fmt.Errorf("hash row %d: %w", row.ID, err)
}
existingHash, exists := existingHashes[row.ID]
if exists && existingHash == hash {
atomic.AddInt64(&unchanged, 1)
return nil
}
idsMu.Lock()
if exists {
updatedIDs = append(updatedIDs, row.ID)
} else {
createdIDs = append(createdIDs, row.ID)
}
idsMu.Unlock()
stampUpdatedAt(row.Doc, runAt)
stampRowHash(row.Doc, hash)
idx := atomic.AddInt64(&changedCount, 1) - 1
changed[idx] = row
return nil
})
}
if err := g.Wait(); err != nil {
result.Err = err
s.lastResult = result
return result.Err
}
changed = changed[:changedCount]
// Antes de escribir hay que leer el estado anterior de las filas que se van a actualizar,
// porque después del BulkUpsert ya no está disponible para el log de cambios.
var beforeDocs map[int]map[string]interface{}
if s.changeLog != nil && s.fetcher != nil && len(updatedIDs) > 0 {
beforeDocs, err = s.fetcher.FindByIDs(ctx, updatedIDs)
if err != nil {
result.Err = fmt.Errorf("fetch previous docs: %w", err)
s.lastResult = result
return result.Err
}
}
// Fase 2 (I/O, un solo round trip): escribir todas las filas que cambiaron en un único BulkWrite.
created, updated, err := s.upserter.BulkUpsert(ctx, changed)
if err != nil {
result.Err = fmt.Errorf("bulk upsert: %w", err)
s.lastResult = result
return result.Err
}
sort.Ints(createdIDs)
sort.Ints(updatedIDs)
s.logChangedIDs(createdIDs, updatedIDs)
s.writeChangeLog(changed, createdIDs, updatedIDs, beforeDocs)
result.Created = created
result.Updated = updated
result.Unchanged = int(unchanged)
result.RowsSynced = len(rows)
deleted, err := s.reconcileDeleted(ctx, pgIDs, existingHashes)
if err != nil {
result.Err = fmt.Errorf("reconcile deleted: %w", err)
s.lastResult = result
return result.Err
}
result.Deleted = deleted
if err := s.state.Set(ctx, runAt); err != nil {
result.Err = fmt.Errorf("advance sync state: %w", err)
s.lastResult = result
return result.Err
}
s.lastResult = result
return nil
}
// logChangedIDs imprime en consola los ids creados y actualizados del ciclo.
func (s *runSyncPipeline) logChangedIDs(createdIDs, updatedIDs []int) {
log.Printf("[%s] created=%d updated=%d", s.name, len(createdIDs), len(updatedIDs))
for _, id := range createdIDs {
log.Printf("[%s] created id=%d", s.name, id)
}
for _, id := range updatedIDs {
log.Printf("[%s] updated id=%d", s.name, id)
}
}
// writeChangeLog vuelca al archivo de cambios el JSON insertado de cada alta y el par
// anterior/nuevo de cada actualización. Un fallo de escritura se registra pero no corta el ciclo.
func (s *runSyncPipeline) writeChangeLog(changed []entities.SyncRow, createdIDs, updatedIDs []int, beforeDocs map[int]map[string]interface{}) {
if s.changeLog == nil {
return
}
docsByID := make(map[int]interface{}, len(changed))
for _, row := range changed {
docsByID[row.ID] = row.Doc
}
for _, id := range createdIDs {
if err := s.changeLog.Insert(id, docsByID[id]); err != nil {
log.Printf("[%s] change log insert id=%d failed: %v", s.name, id, err)
}
}
for _, id := range updatedIDs {
if err := s.changeLog.Update(id, beforeDocs[id], docsByID[id]); err != nil {
log.Printf("[%s] change log update id=%d failed: %v", s.name, id, err)
}
}
}
// reconcileDeleted encuentra ids en Mongo (existingHashes) que ya no están en Postgres y los archiva.
func (s *runSyncPipeline) reconcileDeleted(ctx context.Context, pgIDs map[int]struct{}, existingHashes map[int]string) (int, error) {
if s.lister == nil || s.archiver == nil {
return 0, nil
}
var toDelete []int
for id := range existingHashes {
if _, ok := pgIDs[id]; !ok {
toDelete = append(toDelete, id)
}
}
if len(toDelete) == 0 {
return 0, nil
}
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(s.concurrency)
for _, id := range toDelete {
id := id
g.Go(func() error {
if err := s.archiver.MoveToDeleted(gctx, id); err != nil {
return fmt.Errorf("move record %d to deleted: %w", id, err)
}
return nil
})
}
if err := g.Wait(); err != nil {
return 0, err
}
return len(toDelete), nil
}
// SeedLastResult inicializa lastResult con un timestamp persistido tras arrancar el proceso.
func (s *runSyncPipeline) SeedLastResult(t time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastResult = entities.SyncResult{RanAt: t}
}
// LastResult devuelve una copia del último resultado, protegida por el mutex.
func (s *runSyncPipeline) LastResult() entities.SyncResult {
s.mu.Lock()
defer s.mu.Unlock()
return s.lastResult
}
pkg/changelog/changelog.go 0 → 100644
// Package changelog escribe un archivo de texto con el detalle de cada cambio aplicado
// por un ciclo de sincronización (JSON insertado, o JSON anterior/nuevo en las actualizaciones).
package changelog
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sync"
"time"
"go.mongodb.org/mongo-driver/bson"
)
// DefaultDir es la carpeta donde se dejan los archivos de log de cambios.
const DefaultDir = "/opt/logs"
// Writer agrega entradas al archivo <dir>/sync_<name>.log de un pipeline.
type Writer struct {
mu sync.Mutex
path string
}
// New crea el Writer de un pipeline, asegurando que la carpeta exista.
func New(dir, name string) (*Writer, error) {
if dir == "" {
dir = DefaultDir
}
if err := os.MkdirAll(dir, 0o755); err != nil {
return nil, fmt.Errorf("create log dir %s: %w", dir, err)
}
return &Writer{path: filepath.Join(dir, fmt.Sprintf("sync_%s.log", name))}, nil
}
// Path devuelve la ruta del archivo de log.
func (w *Writer) Path() string {
return w.path
}
// Insert registra el documento insertado.
func (w *Writer) Insert(id int, doc interface{}) error {
return w.write(fmt.Sprintf("INSERT id=%d\n new: %s\n", id, marshal(doc)))
}
// Update registra el documento anterior y el nuevo.
func (w *Writer) Update(id int, before, after interface{}) error {
return w.write(fmt.Sprintf("UPDATE id=%d\n old: %s\n new: %s\n", id, marshal(before), marshal(after)))
}
// write agrega una entrada con su timestamp al archivo, en modo append.
func (w *Writer) write(entry string) error {
w.mu.Lock()
defer w.mu.Unlock()
f, err := os.OpenFile(w.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("open log %s: %w", w.path, err)
}
defer f.Close()
_, err = fmt.Fprintf(f, "[%s] %s", time.Now().UTC().Format(time.RFC3339), entry)
return err
}
// marshal serializa a JSON en una línea usando los MISMOS nombres de campo que la colección Mongo.
// Las entidades tienen tags bson (no json), así que se pasa por un round trip bson -> map para que
// el documento nuevo y el anterior (leído de Mongo) sean comparables campo a campo con cualquier diff.
// Si falla, devuelve el error como texto para no perder la entrada.
func marshal(v interface{}) string {
doc, err := toCollectionShape(v)
if err != nil {
return fmt.Sprintf("<marshal error: %v>", err)
}
b, err := json.Marshal(doc)
if err != nil {
return fmt.Sprintf("<marshal error: %v>", err)
}
return string(b)
}
// toCollectionShape normaliza cualquier documento (entidad tipada o bson.M leído de Mongo)
// a un map con los nombres de campo de la colección, descartando `_id` (interno de Mongo).
func toCollectionShape(v interface{}) (map[string]interface{}, error) {
if v == nil {
return nil, nil
}
b, err := bson.Marshal(v)
if err != nil {
return nil, err
}
var doc map[string]interface{}
if err := bson.Unmarshal(b, &doc); err != nil {
return nil, err
}
delete(doc, "_id")
return doc, nil
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment