nexarch/connect/internal/masterdatasync/masterdatasync.go

// Package masterdatasync implements ERP-03: automated reconciliation of
// customer/supplier master data between NEXARCH and a connected ERP/CRM
// system, on top of ERP-01/ERP-02's connector clients and ENG-03's
// syncjob schedule.
//
// AC1: a record new on the external side is inserted and thus visible in
// NEXARCH after Reconcile runs.
// AC2: if both the external record and the NEXARCH-side copy changed since
// the last sync, Reconcile marks the record conflicted instead of silently
// picking a winner — the external change is not applied.
// AC3: RunDueJobs drives Reconcile from ENG-03's syncjob schedule (ClaimDue/
// CompleteRun), so a scheduled run and a manual TriggerNow both execute the
// same reconciliation logic.
package masterdatasync

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/connect/internal/syncjob"
)

var ErrNotFound = errors.New("masterdatasync: not found")

// Ref identifies the connector instance a set of master-data records
// belongs to.
type Ref struct {
	TenantSlug          string
	ConnectorName       string
	ConnectorInstanceID string
}

// Entity is the kind of master-data record being synced.
type Entity string

const (
	EntityKunde     Entity = "kunde"
	EntityLieferant Entity = "lieferant"
)

// ExternalRecord is one record as read from the ERP/CRM system: an
// external ID plus its raw field data.
type ExternalRecord struct {
	ExternalID string
	Data       json.RawMessage
}

// Source is the minimal capability Reconcile needs from a connector: list
// every record of an entity. erpconnector.Client and crmconnector.Client
// both implement this via their List method.
type Source interface {
	List(ctx context.Context, entity string) ([]byte, error)
}

// Record is the stored reconciliation state for one external record.
type Record struct {
	ID                 int64
	Ref                Ref
	Entity             Entity
	ExternalID         string
	ExternalData       json.RawMessage
	NexarchData        json.RawMessage
	Conflict           bool
	ConflictDetectedAt *time.Time
}

// Result summarizes one Reconcile call.
type Result struct {
	Created   int
	Updated   int
	Conflicts int
}

// Store persists master-data reconciliation state in Postgres.
type Store struct {
	pool *pgxpool.Pool
}

func NewStore(pool *pgxpool.Pool) *Store {
	return &Store{pool: pool}
}

// canonicalize re-encodes data through encoding/json (map keys sorted,
// whitespace normalized) so a value hashes identically regardless of
// whether it just came from a caller, was re-marshaled, or round-tripped
// through Postgres JSONB (which reformats whitespace on its own).
func canonicalize(data json.RawMessage) (json.RawMessage, error) {
	var v any
	if err := json.Unmarshal(data, &v); err != nil {
		return nil, fmt.Errorf("canonicalize: %w", err)
	}
	out, err := json.Marshal(v)
	if err != nil {
		return nil, fmt.Errorf("canonicalize: %w", err)
	}
	return out, nil
}

func hashOf(data json.RawMessage) string {
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:])
}

// Get returns the stored record for one external ID, or ErrNotFound.
func (s *Store) Get(ctx context.Context, ref Ref, entity Entity, externalID string) (Record, error) {
	var r Record
	var conflictAt *time.Time
	err := s.pool.QueryRow(ctx, `
		SELECT id, tenant_slug, connector_name, connector_instance_id, entity, external_id,
		       external_data, nexarch_data, conflict, conflict_detected_at
		FROM master_data_records
		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
		  AND entity = $4 AND external_id = $5
	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity), externalID).Scan(
		&r.ID, &r.Ref.TenantSlug, &r.Ref.ConnectorName, &r.Ref.ConnectorInstanceID, &r.Entity, &r.ExternalID,
		&r.ExternalData, &r.NexarchData, &r.Conflict, &conflictAt,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return Record{}, ErrNotFound
	}
	if err != nil {
		return Record{}, fmt.Errorf("masterdatasync: get: %w", err)
	}
	r.ConflictDetectedAt = conflictAt
	return r, nil
}

// ListByRef returns every stored record for a connector instance and
// entity, for inspection/testing.
func (s *Store) ListByRef(ctx context.Context, ref Ref, entity Entity) ([]Record, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT id, tenant_slug, connector_name, connector_instance_id, entity, external_id,
		       external_data, nexarch_data, conflict, conflict_detected_at
		FROM master_data_records
		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND entity = $4
		ORDER BY external_id
	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity))
	if err != nil {
		return nil, fmt.Errorf("masterdatasync: list: %w", err)
	}
	defer rows.Close()
	var out []Record
	for rows.Next() {
		var r Record
		var conflictAt *time.Time
		if err := rows.Scan(&r.ID, &r.Ref.TenantSlug, &r.Ref.ConnectorName, &r.Ref.ConnectorInstanceID, &r.Entity,
			&r.ExternalID, &r.ExternalData, &r.NexarchData, &r.Conflict, &conflictAt); err != nil {
			return nil, fmt.Errorf("masterdatasync: list: scan: %w", err)
		}
		r.ConflictDetectedAt = conflictAt
		out = append(out, r)
	}
	return out, rows.Err()
}

// SetLocalData simulates/records a NEXARCH-side edit to a record, made
// independently of any sync. It exists so a local change can be present
// the next time Reconcile runs and be compared against a concurrent
// external change (AC2). It does not touch external_data.
func (s *Store) SetLocalData(ctx context.Context, ref Ref, entity Entity, externalID string, data json.RawMessage) error {
	canon, err := canonicalize(data)
	if err != nil {
		return fmt.Errorf("masterdatasync: set local data: %w", err)
	}
	tag, err := s.pool.Exec(ctx, `
		UPDATE master_data_records SET nexarch_data = $6, updated_at = now()
		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
		  AND entity = $4 AND external_id = $5
	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity), externalID, canon)
	if err != nil {
		return fmt.Errorf("masterdatasync: set local data: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

// ResolveConflict clears a record's conflict flag and adopts resolved as
// the new NEXARCH-side state, so the next Reconcile call no longer treats
// it as diverged. Never called automatically — a human or an ERP-04
// process decides the resolution.
func (s *Store) ResolveConflict(ctx context.Context, ref Ref, entity Entity, externalID string, resolved json.RawMessage) error {
	canon, err := canonicalize(resolved)
	if err != nil {
		return fmt.Errorf("masterdatasync: resolve conflict: %w", err)
	}
	tag, err := s.pool.Exec(ctx, `
		UPDATE master_data_records
		SET nexarch_data = $6, nexarch_hash = $7, conflict = false, conflict_detected_at = NULL, updated_at = now()
		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
		  AND entity = $4 AND external_id = $5
	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity), externalID, canon, hashOf(canon))
	if err != nil {
		return fmt.Errorf("masterdatasync: resolve conflict: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

// Reconcile fetches every record of entity from source and reconciles it
// against stored state:
//   - unseen external_id -> inserted as a new NEXARCH record (AC1).
//   - external side changed, NEXARCH side unchanged since last sync ->
//     the external change is adopted.
//   - external side changed AND the NEXARCH side was independently edited
//     (via SetLocalData) since the last sync -> flagged as a conflict, the
//     external change is NOT applied (AC2).
//   - neither side changed -> no-op.
func (s *Store) Reconcile(ctx context.Context, ref Ref, entity Entity, source Source) (Result, error) {
	raw, err := source.List(ctx, string(entity))
	if err != nil {
		return Result{}, fmt.Errorf("masterdatasync: reconcile: list: %w", err)
	}
	var records []ExternalRecord
	if err := json.Unmarshal(raw, &records); err != nil {
		return Result{}, fmt.Errorf("masterdatasync: reconcile: decode: %w", err)
	}

	var res Result
	for _, ext := range records {
		outcome, err := s.reconcileOne(ctx, ref, entity, ext)
		if err != nil {
			return res, fmt.Errorf("masterdatasync: reconcile %s: %w", ext.ExternalID, err)
		}
		switch outcome {
		case outcomeCreated:
			res.Created++
		case outcomeConflicted:
			res.Conflicts++
		case outcomeUpdated:
			res.Updated++
		}
	}
	return res, nil
}

type reconcileOutcome int

const (
	outcomeNoop reconcileOutcome = iota
	outcomeCreated
	outcomeUpdated
	outcomeConflicted
)

func (s *Store) reconcileOne(ctx context.Context, ref Ref, entity Entity, ext ExternalRecord) (outcome reconcileOutcome, err error) {
	canonExternal, err := canonicalize(ext.Data)
	if err != nil {
		return outcomeNoop, fmt.Errorf("canonicalize external data: %w", err)
	}
	newExternalHash := hashOf(canonExternal)

	var existingID int64
	var storedExternalHash, storedNexarchHash string
	var nexarchData json.RawMessage
	err = s.pool.QueryRow(ctx, `
		SELECT id, external_hash, nexarch_hash, nexarch_data FROM master_data_records
		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
		  AND entity = $4 AND external_id = $5
		FOR UPDATE
	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity), ext.ExternalID).
		Scan(&existingID, &storedExternalHash, &storedNexarchHash, &nexarchData)

	if errors.Is(err, pgx.ErrNoRows) {
		_, insertErr := s.pool.Exec(ctx, `
			INSERT INTO master_data_records
				(tenant_slug, connector_name, connector_instance_id, entity, external_id,
				 external_data, external_hash, nexarch_data, nexarch_hash)
			VALUES ($1, $2, $3, $4, $5, $6, $7, $6, $7)
		`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, string(entity), ext.ExternalID,
			canonExternal, newExternalHash)
		if insertErr != nil {
			return outcomeNoop, fmt.Errorf("insert: %w", insertErr)
		}
		return outcomeCreated, nil
	}
	if err != nil {
		return outcomeNoop, fmt.Errorf("lookup: %w", err)
	}

	canonNexarch, err := canonicalize(nexarchData)
	if err != nil {
		return outcomeNoop, fmt.Errorf("canonicalize stored nexarch data: %w", err)
	}

	externalChanged := storedExternalHash != newExternalHash
	localChanged := hashOf(canonNexarch) != storedNexarchHash
	if !externalChanged {
		return outcomeNoop, nil
	}
	if localChanged {
		_, err = s.pool.Exec(ctx, `
			UPDATE master_data_records
			SET external_data = $2, external_hash = $3, conflict = true, conflict_detected_at = now(), updated_at = now()
			WHERE id = $1
		`, existingID, canonExternal, newExternalHash)
		if err != nil {
			return outcomeNoop, fmt.Errorf("mark conflict: %w", err)
		}
		return outcomeConflicted, nil
	}

	_, err = s.pool.Exec(ctx, `
		UPDATE master_data_records
		SET external_data = $2, external_hash = $3, nexarch_data = $2, nexarch_hash = $3, updated_at = now()
		WHERE id = $1
	`, existingID, canonExternal, newExternalHash)
	if err != nil {
		return outcomeNoop, fmt.Errorf("adopt external change: %w", err)
	}
	return outcomeUpdated, nil
}

// Runner drives Reconcile from ENG-03's syncjob schedule (AC3): a due or
// manually triggered run performs the same reconciliation as a direct
// Reconcile call, and its outcome (success/failure) is recorded on the
// syncjob run so scheduled and manual runs are equally observable.
type Runner struct {
	Jobs   *syncjob.Store
	Store  *Store
	Source Source
	Entity Entity
}

// RunDueJobs claims every syncjob due at now (ENG-03's syncjob.ClaimDue
// sweeps the shared sync_jobs table, not just this Runner's own job) and
// reconciles the one matching jobID, completing its run with the outcome.
// Building block for a future background worker that dispatches every
// claimed run to its connector by looking up the job's Ref (none exists
// yet, same documented-limitation pattern as connect's other ClaimDue-style
// functions this session, e.g. datevtransfer.ClaimDueRetries) — runs for
// other jobs claimed in the same sweep are left running until a real
// dispatcher completes them.
func (r *Runner) RunDueJobs(ctx context.Context, ref Ref, jobID string, now time.Time) (int, error) {
	runs, err := r.Jobs.ClaimDue(ctx, now)
	if err != nil {
		return 0, fmt.Errorf("masterdatasync: run due jobs: claim: %w", err)
	}
	n := 0
	for _, run := range runs {
		if run.SyncJobID != jobID {
			continue
		}
		_, reconcileErr := r.Store.Reconcile(ctx, ref, r.Entity, r.Source)
		if completeErr := r.Jobs.CompleteRun(ctx, run.ID, reconcileErr); completeErr != nil {
			return n, fmt.Errorf("masterdatasync: run due jobs: complete run: %w", completeErr)
		}
		n++
	}
	return n, nil
}