363 lines
13 KiB
Go
363 lines
13 KiB
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
|
|
}
|