feat: agregar plan de pago a estudiantes y limpiar reader

Añade entidad payment_plan con su reader dedicado y simplifica
student_reader delegando la lectura del plan de pago.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NkYn6CP4LmjWA5cturKj8R

internal/domain/entities/payment_plan.go

+package entities
+
+import "time"
+
+type PaymentPlan struct {
+	PaymentPlanID      int       `bson:"payment_plan_id"`
+	StudentID          int       `bson:"student_id"`
+	EnrollmentID       int       `bson:"enrollment_id"`
+	Order              int       `bson:"payment_plan_order"`
+	Concept            *string   `bson:"payment_plan_concept"`
+	Year               int       `bson:"payment_plan_year"`
+	Date               *string   `bson:"payment_plan_date"`
+	AdditionalConcepts *string   `bson:"payment_plan_additional_concepts"`
+	IssueDate          *string   `bson:"payment_plan_issue_date"`
+	Total              *float64  `bson:"payment_plan_total"`
+	Debt               *int      `bson:"payment_plan_debt"`
+	CreatedAt          time.Time `bson:"created_at"`
+	UpdatedAt          time.Time `bson:"updated_at"`
+	RowHash            string    `bson:"row_hash,omitempty" json:"-"`
+}
+
+func (p *PaymentPlan) SetUpdatedAt(t time.Time) {
+	p.UpdatedAt = t
+}
+
+func (p *PaymentPlan) SetRowHash(h string) {
+	p.RowHash = h
+}
+
+func (p *PaymentPlan) SetCreatedAt(t time.Time) {
+	p.CreatedAt = t
+}

internal/domain/entities/student.go

@@ -19,29 +19,12 @@ type Student struct {
 	ClassroomAttendanceID *int       `bson:"student_classroom_attendance_id"`
 	PeriodID              int        `bson:"student_period_id"`
 	CicleID               int        `bson:"student_cicle_id"`
-	PaymentPlans          []PaymentPlanYear `bson:"payment_plans"`
+	EnrollmentID          int        `bson:"enrollment_id"`
 	CreatedAt             time.Time  `bson:"created_at"`
 	UpdatedAt             time.Time  `bson:"updated_at"`
 	RowHash               string     `bson:"row_hash,omitempty" json:"-"`
 }
 
-type PaymentPlanYear struct {
-	Year     int       `bson:"year"`
-	Payments []Payment `bson:"payments"`
-}
-
-type Payment struct {
-	PaymentPlanID      int      `bson:"payment_plan_id"`
-	EnrollmentID       int      `bson:"enrollment_id"`
-	Order              int      `bson:"payment_plan_order"`
-	Concept            *string  `bson:"payment_plan_concept"`
-	Date               *string  `bson:"payment_plan_date"`
-	AdditionalConcepts *string  `bson:"payment_plan_additional_concepts"`
-	IssueDate          *string  `bson:"payment_plan_issue_date"`
-	Total              *float64 `bson:"payment_plan_total"`
-	Debt               *int     `bson:"payment_plan_debt"`
-}
-
 func (s *Student) SetUpdatedAt(t time.Time) {
 	s.UpdatedAt = t
 }

internal/infrastructure/config/pipelines.go

@@ -21,6 +21,14 @@ var Pipelines = []PipelineDef{
 		StateID:                1,
 	},
 	{
+		Name:                   "payment_plans",
+		PGQuery:                postgres.PaymentPlanQuery,
+		MongoCollection:        "payment_plans",
+		MongoDeletedCollection: "deleted_payment_plans",
+		IDField:                "payment_plan_id",
+		StateID:                2,
+	},
+	{
 		Name:                   "parents",
 		PGQuery:                postgres.ParentQuery,
 		MongoCollection:        "parents",

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

internal/infrastructure/database/postgres/payment_plan_reader.go

+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"intranet-synchronizer/internal/domain/entities"
+	"intranet-synchronizer/internal/domain/repositories"
+)
+
+type PaymentPlanReader struct {
+	pool  *Pool
+	query string
+}
+
+func NewPaymentPlanReader(pool *Pool, query string) repositories.SourceReader {
+	return &PaymentPlanReader{pool: pool, query: query}
+}
+
+func scanPaymentPlanRow(row pgx.CollectableRow) (entities.SyncRow, error) {
+	var doc entities.PaymentPlan
+	if err := row.Scan(
+		&doc.PaymentPlanID,
+		&doc.StudentID,
+		&doc.EnrollmentID,
+		&doc.Order,
+		&doc.Concept,
+		&doc.Year,
+		&doc.Date,
+		&doc.AdditionalConcepts,
+		&doc.IssueDate,
+		&doc.Total,
+		&doc.Debt,
+	); err != nil {
+		return entities.SyncRow{}, fmt.Errorf("scan payment_plan row: %w", err)
+	}
+	return entities.SyncRow{ID: doc.PaymentPlanID, Doc: &doc}, nil
+}
+
+func (f *PaymentPlanReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error) {
+	rows, err := f.pool.Pool.Query(ctx, f.query)
+	if err != nil {
+		return nil, fmt.Errorf("query payment_plans: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, scanPaymentPlanRow)
+	if err != nil {
+		return nil, fmt.Errorf("iterate payment_plan rows: %w", err)
+	}
+	return result, nil
+}

internal/infrastructure/database/postgres/student_query.go

 package postgres
 
 // 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.
+// Los planes de pago son una tubería aparte (payment_plans) — no se anidan en el documento de estudiante.
 const StudentQuery = `
 SELECT e.estudiante_id                                   as student_id,
        e.estudiante_codigo                               as student_code,
@@ -20,7 +18,8 @@ SELECT e.estudiante_id                                   as student_id,
        public.to_camel_case(grado.grado_sigla)           as student_grade,
        ma.aula_id_asiste                                 as student_classroom_attendance_id,
        14                                                 as student_period_id,
-       28                                                 as student_cicle_id
+       28                                                 as student_cicle_id,
+       ma.matricula_id									  as enrollment_id
 FROM matricula.ma_estudiante e
          INNER JOIN persona.pe_persona pp ON e.persona_id = pp.persona_id
          INNER JOIN general.gn_catalogo_siiaa sexo on pp.sexo_id = sexo.catalogo_siiaa_id

internal/infrastructure/database/postgres/student_reader.go

@@ -3,7 +3,6 @@ package postgres
 import (
 	"context"
 	"fmt"
-	"sort"
 	"time"
 
 	"intranet-synchronizer/internal/domain/entities"
@@ -25,16 +24,9 @@ func (f *StudentReader) FetchAll(ctx context.Context) ([]entities.SyncRow, error
 		return nil, err
 	}
 
-	plansByStudent, err := f.fetchPaymentPlans(ctx)
-	if err != nil {
-		return nil, err
-	}
-
 	result := make([]entities.SyncRow, 0, len(order))
 	for _, studentID := range order {
-		doc := students[studentID]
-		doc.PaymentPlans = buildPaymentPlanYears(plansByStudent[studentID])
-		result = append(result, entities.SyncRow{ID: studentID, Doc: doc})
+		result = append(result, entities.SyncRow{ID: studentID, Doc: students[studentID]})
 	}
 	return result, nil
 }
@@ -66,10 +58,11 @@ func FetchStudentBasics(ctx context.Context, pool *Pool, query string) (map[int]
 			classroomAttendanceID *int
 			periodID              int
 			cicleID               int
+			enrollmentID          int
 		)
 		if err := rows.Scan(&studentID, &studentCode, &studentInternalCode, &paternalLastName, &maternalLastName,
 			&name, &dni, &birthday, &email, &genre, &branch, &level,
-			&grade, &classroomAttendanceID, &periodID, &cicleID); err != nil {
+			&grade, &classroomAttendanceID, &periodID, &cicleID, &enrollmentID); err != nil {
 			return nil, nil, fmt.Errorf("scan student row: %w", err)
 		}
 
@@ -90,6 +83,7 @@ func FetchStudentBasics(ctx context.Context, pool *Pool, query string) (map[int]
 			ClassroomAttendanceID: classroomAttendanceID,
 			PeriodID:              periodID,
 			CicleID:               cicleID,
+			EnrollmentID:          enrollmentID,
 		}
 		order = append(order, studentID)
 	}
@@ -98,75 +92,3 @@ func FetchStudentBasics(ctx context.Context, pool *Pool, query string) (map[int]
 	}
 	return students, order, nil
 }
-
-type paymentRow struct {
-	entities.Payment
-	year int
-}
-
-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)
-	}
-	defer rows.Close()
-
-	plansByStudent := make(map[int][]paymentRow)
-	for rows.Next() {
-		var (
-			paymentPlanID      int
-			studentID          int
-			enrollmentID       int
-			order              int
-			concept            *string
-			year               int
-			date               *string
-			additionalConcepts *string
-			issueDate          *string
-			total              *float64
-			debt               *int
-		)
-		if err := rows.Scan(&paymentPlanID, &studentID, &enrollmentID, &order, &concept, &year,
-			&date, &additionalConcepts, &issueDate, &total, &debt); err != nil {
-			return nil, fmt.Errorf("scan payment_plan row: %w", err)
-		}
-
-		plansByStudent[studentID] = append(plansByStudent[studentID], paymentRow{
-			Payment: entities.Payment{
-				PaymentPlanID:      paymentPlanID,
-				EnrollmentID:       enrollmentID,
-				Order:              order,
-				Concept:            concept,
-				Date:               date,
-				AdditionalConcepts: additionalConcepts,
-				IssueDate:          issueDate,
-				Total:              total,
-				Debt:               debt,
-			},
-			year: year,
-		})
-	}
-	if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("iterate payment_plan rows: %w", err)
-	}
-	return plansByStudent, nil
-}
-
-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)
-	}
-
-	years := make([]int, 0, len(byYear))
-	for y := range byYear {
-		years = append(years, y)
-	}
-	sort.Sort(sort.Reverse(sort.IntSlice(years)))
-
-	result := make([]entities.PaymentPlanYear, 0, len(years))
-	for _, y := range years {
-		result = append(result, entities.PaymentPlanYear{Year: y, Payments: byYear[y]})
-	}
-	return result
-}

internal/infrastructure/pipeline/builder.go

@@ -12,13 +12,14 @@ import (
 	"intranet-synchronizer/internal/infrastructure/database/mongo/repository"
 	"intranet-synchronizer/internal/infrastructure/database/postgres"
 	"intranet-synchronizer/internal/usecase"
-	"intranet-synchronizer/pkg/changelog"
 )
 
 func readerFor(name string, pool *postgres.Pool, query string) (repositories.SourceReader, error) {
 	switch name {
 	case "students":
 		return postgres.NewStudentReader(pool, query), nil
+	case "payment_plans":
+		return postgres.NewPaymentPlanReader(pool, query), nil
 	case "parents":
 		return postgres.NewParentReader(pool, query), nil
 	case "users":
@@ -83,10 +84,14 @@ func Build(
 		archiver := repository.NewDeletedRepository(db.Collection(p.MongoCollection), db.Collection(p.MongoDeletedCollection), p.IDField)
 		state := postgres.NewSyncStateRepository(pool, p.StateID)
 
-		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)
-		}
+		//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)
+		//}
+
+		// Escritura del log de cambios deshabilitada por el momento: con changeLog nil el caso de uso
+		// omite el fetch de documentos previos y no escribe en /opt/logs.
+		var changeLog usecase.ChangeLogger
 
 		uc := usecase.NewRunSyncPipelineUseCase(p.Name, reader, docRepo, docRepo, archiver, state, docRepo, changeLog, time.Now)