M365-02: SharePoint/OneDrive Dokumentabgleich #241
No files matched your search
@@ -0,0 +1,40 @@
|
||||
# M365-02 – Prüfprotokoll: SharePoint/OneDrive-Dokumentabgleich
|
||||
|
||||
Datum: 2026-09-28. Host: Entwicklungshost 192.168.1.204, eigene Wegwerf-DB/-Rolle `m365_02_test` (angelegt für diesen Testlauf, keine Kollision mit anderen parallelen Agenten). Frischer Klon von `origin/main` in eigenem Wegwerf-Arbeitsverzeichnis.
|
||||
|
||||
## Umfang
|
||||
|
||||
- `connect/internal/m365docsync`: Reconciliation von Dokumenten zwischen NEXARCH und einem SharePoint/OneDrive-Laufwerk über Microsoft Graph (M365-01, `m365connector.Client.Read`), auf demselben Hash-Vergleichs-Muster wie `masterdatasync` (ERP-03) — Microsoft 365 ist ein Connector unter demselben Sync-Modell, keine Sonderbehandlung.
|
||||
- `Store.Reconcile`: neues Dokument -> Insert mit Metadaten (Name, Pfad, Größe, Änderungszeitpunkt); nur extern geändert -> übernommen; extern **und** lokal geändert -> als Konflikt markiert, externe Änderung **nicht** angewendet.
|
||||
- `GraphSource`: adaptiert Microsoft Graphs Kind-Element-Listing (`GET /drives/{id}/root/children`) zu `ExternalDocument`, überspringt Ordner, nutzt `file.hashes.quickXorHash` als Änderungsmarker (Fallback auf Größe+Zeitstempel falls Graph noch keinen Hash berechnet hat).
|
||||
- `Runner.RunDueJobs`: treibt `Reconcile` über die bestehende Sync-Job-Verwaltung (`syncjob.Store.ClaimDue`/`CompleteRun`, ENG-03) an — ein geplanter und ein manuell ausgelöster Lauf durchlaufen dieselbe Logik.
|
||||
- Migration `0011_m365_doc_records`: Tabelle `m365_doc_records` mit getrennten Hash-Spalten für zuletzt von Graph gesehenen Stand (`external_hash`) und zuletzt NEXARCH-seitig übernommenen Stand (`nexarch_hash`), plus `conflict`/`conflict_detected_at`.
|
||||
- `scripts/check-security-test-coverage.sh`: `internal/m365docsync` in die Liste sicherheitsrelevanter Pakete mit Testpflicht aufgenommen (Tenant-gescopte Daten).
|
||||
|
||||
## Limitation: kein echter Microsoft-Testmandant
|
||||
|
||||
Wie schon bei M365-01 dokumentiert stehen in dieser Umgebung keine echten Azure-AD/Microsoft-365-Testmandanten-Credentials zur Verfügung. Der Graph-Aufruf für die Kind-Element-Listing wird gegen einen `httptest.Server`-Mock geprüft, der den realen Antwortvertrag von `GET /drives/{id}/root/children` nachbildet (Graph-Collection `{"value":[...]}`, Datei- vs. Ordner-Facette, `file.hashes.quickXorHash`). Ein echter End-to-End-Lauf gegen einen produktiven Microsoft-Testmandanten bzw. eine echte SharePoint-Site ist damit **offen** und ehrlich als solcher dokumentiert, nicht übersprungen.
|
||||
|
||||
## Akzeptanzkriterien
|
||||
|
||||
| # | Kriterium | Ergebnis |
|
||||
|---|---|---|
|
||||
| 1 | Dokument aus SharePoint/OneDrive erscheint nach Sync in NEXARCH mit Metadaten | **Erfüllt.** `TestReconcile_NewExternalDocument_IsCreatedWithMetadata` (Name, Pfad, Größe korrekt übernommen) + `TestGraphSource_ListDocuments_ParsesChildrenListing` (echter HTTP-Mock-Aufruf gegen die Graph-Kind-Element-Listing, Ordner werden übersprungen). |
|
||||
| 2 | Änderungskonflikt wird erkannt und nicht stillschweigend überschrieben | **Erfüllt.** `TestReconcile_ConcurrentChangeBothSides_IsFlaggedNotOverwritten` — gleichzeitige externe und lokale Änderung wird als Konflikt markiert, `nexarch_hash` bleibt beim lokalen Stand statt überschrieben zu werden. `TestResolveConflict_ClearsFlagAndAdoptsResolvedHash` als Gegenprobe für die manuelle Auflösung. |
|
||||
| 3 | Abgleich läuft nach Zeitplan über die Sync-Job-Verwaltung | **Erfüllt.** `TestRunner_RunDueJobs_ExecutesReconcileOnSchedule` — `Runner.RunDueJobs` claimt einen fälligen `syncjob`-Lauf und führt darüber `Reconcile` aus, Lauf wird als `succeeded` protokolliert. |
|
||||
|
||||
## Prüfungen vor der Abnahme
|
||||
|
||||
1. **Testdokument aus SharePoint nach Sync in NEXARCH korrekt vorhanden.** Zielwert: nach `Reconcile` ist ein neues Drive-Item als `m365_doc_records`-Zeile mit korrekten Metadaten (Name, Pfad, Größe) lesbar. Ergebnis: **bestanden** — `TestReconcile_NewExternalDocument_IsCreatedWithMetadata`, `TestReconcile_NoChange_IsNoop` (Gegenprobe: kein doppelter Insert bei unverändertem Stand).
|
||||
2. **Gleichzeitige Änderung auf beiden Seiten erzeugt erkennbaren Konflikt.** Zielwert: `conflict = true`, externe Änderung wird nicht angewendet (NEXARCH-Hash bleibt unverändert), aber der neue externe Stand wird zur späteren Auflösung mitgetrackt. Ergebnis: **bestanden** — `TestReconcile_ConcurrentChangeBothSides_IsFlaggedNotOverwritten`.
|
||||
3. **Zeitplan-gesteuerter Lauf im Test erfolgreich.** Zielwert: ein über `syncjob.Store.Create` angelegter, fälliger Job wird von `Runner.RunDueJobs` geclaimt, führt zu einem tatsächlich ausgeführten `Reconcile`-Lauf und wird als `succeeded` abgeschlossen. Ergebnis: **bestanden** — `TestRunner_RunDueJobs_ExecutesReconcileOnSchedule`.
|
||||
|
||||
## Regressionsprüfung
|
||||
|
||||
`go build ./...`, `go vet ./...` für das gesamte `connect`-Modul: keine Fehler. `go test ./internal/m365docsync/... -v -p 1` (gegen `TEST_TENANT_DSN` der Wegwerf-DB `m365_02_test`): alle 7 Tests grün. `scripts/check-security-test-coverage.sh`: Merge-Gate bestanden, `internal/m365docsync` mit Testpflicht erfasst.
|
||||
|
||||
## Nicht Bestandteil dieser Kachel (nicht umgesetzt)
|
||||
|
||||
- `QA-03` (Prüfgate ERP/DATEV/M365) — laut Ticket ausdrücklich eine andere Kachel; die Schnittstelle (`Store`, `Runner`, klar typisierte Fehler) ist so gehalten, dass ein späteres Prüfgate ohne Umbau darauf aufbauen kann.
|
||||
- Automatischer Hintergrund-Dispatcher, der `syncjob.ClaimDue`-Ergebnisse über alle Connector-Instanzen hinweg verteilt (gleiche dokumentierte Lücke wie bei `masterdatasync.Runner` und `erpexport.Store.ClaimDueRetries` — `RunDueJobs` ist der Baustein dafür, kein eigener Worker-Prozess existiert bisher in diesem Modul).
|
||||
- Interaktive Konfliktauflösungs-UI — `ResolveConflict` ist die Store-Primitive dafür, keine Oberfläche ist Teil dieser Kachel.
|
||||
@@ -0,0 +1,95 @@
|
||||
package m365docsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/connect/internal/credential"
|
||||
)
|
||||
|
||||
// GraphReader is the minimal capability GraphSource needs from
|
||||
// m365connector.Client: an authenticated raw Graph GET. Kept as an
|
||||
// interface (rather than importing *m365connector.Client directly) so the
|
||||
// dependency stays the same direction as erpexport -> erpconnector.
|
||||
type GraphReader interface {
|
||||
Read(ctx context.Context, secret credential.Secret, path string) ([]byte, error)
|
||||
}
|
||||
|
||||
// GraphSource lists the documents of one SharePoint/OneDrive drive folder
|
||||
// via Microsoft Graph's drive-item "children" listing, and adapts them to
|
||||
// ExternalDocument for Reconcile.
|
||||
type GraphSource struct {
|
||||
Graph GraphReader
|
||||
Secret credential.Secret
|
||||
DriveID string
|
||||
ItemPath string
|
||||
}
|
||||
|
||||
func (g GraphSource) childrenPath() string {
|
||||
if g.ItemPath == "" || g.ItemPath == "/" {
|
||||
return "/drives/" + g.DriveID + "/root/children"
|
||||
}
|
||||
return "/drives/" + g.DriveID + "/root:" + g.ItemPath + ":/children"
|
||||
}
|
||||
|
||||
type graphDriveItemList struct {
|
||||
Value []graphDriveItem `json:"value"`
|
||||
}
|
||||
|
||||
type graphDriveItem struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Size int64 `json:"size"`
|
||||
LastModifiedDateTime string `json:"lastModifiedDateTime"`
|
||||
ParentReference struct {
|
||||
Path string `json:"path"`
|
||||
} `json:"parentReference"`
|
||||
File *struct {
|
||||
Hashes struct {
|
||||
QuickXorHash string `json:"quickXorHash"`
|
||||
} `json:"hashes"`
|
||||
} `json:"file"`
|
||||
Folder *struct{} `json:"folder"`
|
||||
}
|
||||
|
||||
// ListDocuments implements Source. Folders in the listing are skipped --
|
||||
// only actual drive items with a "file" facet are synced as documents.
|
||||
func (g GraphSource) ListDocuments(ctx context.Context) ([]ExternalDocument, error) {
|
||||
body, err := g.Graph.Read(ctx, g.Secret, g.childrenPath())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("m365docsync: list documents: %w", err)
|
||||
}
|
||||
var list graphDriveItemList
|
||||
if err := json.Unmarshal(body, &list); err != nil {
|
||||
return nil, fmt.Errorf("m365docsync: list documents: decode: %w", err)
|
||||
}
|
||||
|
||||
out := make([]ExternalDocument, 0, len(list.Value))
|
||||
for _, item := range list.Value {
|
||||
if item.File == nil {
|
||||
continue
|
||||
}
|
||||
modifiedAt, err := time.Parse(time.RFC3339, item.LastModifiedDateTime)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("m365docsync: list documents: parse lastModifiedDateTime for %s: %w", item.ID, err)
|
||||
}
|
||||
hash := item.File.Hashes.QuickXorHash
|
||||
if hash == "" {
|
||||
// Fallback for drive items Graph did not compute a hash for
|
||||
// yet -- size+modification time is still enough to detect a
|
||||
// change, just less precise than a content hash.
|
||||
hash = fmt.Sprintf("%d-%s", item.Size, item.LastModifiedDateTime)
|
||||
}
|
||||
out = append(out, ExternalDocument{
|
||||
ExternalID: item.ID,
|
||||
Name: item.Name,
|
||||
Path: item.ParentReference.Path,
|
||||
SizeBytes: item.Size,
|
||||
ModifiedAt: modifiedAt,
|
||||
ContentHash: hash,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,312 @@
|
||||
// Package m365docsync implements M365-02: reconciliation of documents
|
||||
// between NEXARCH and a SharePoint/OneDrive drive over the Microsoft-Graph
|
||||
// connector (M365-01), on top of ENG-03's syncjob schedule. Same
|
||||
// reconcile/hash-comparison shape as masterdatasync (ERP-03) -- Microsoft
|
||||
// 365 is one connector under the same sync model, not a special case.
|
||||
//
|
||||
// AC1: a document new on the SharePoint/OneDrive side is inserted with its
|
||||
// metadata and thus visible in NEXARCH after Reconcile runs.
|
||||
// AC2: if both the drive item 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 m365docsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"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("m365docsync: not found")
|
||||
|
||||
// Ref identifies the connector instance a set of synced documents belongs
|
||||
// to.
|
||||
type Ref struct {
|
||||
TenantSlug string
|
||||
ConnectorName string
|
||||
ConnectorInstanceID string
|
||||
}
|
||||
|
||||
// ExternalDocument is one drive item as read from SharePoint/OneDrive via
|
||||
// Microsoft Graph: its metadata plus a content hash used to detect whether
|
||||
// it changed since the last sync (Graph's file.hashes.quickXorHash, or any
|
||||
// other opaque version marker the Source chooses).
|
||||
type ExternalDocument struct {
|
||||
ExternalID string
|
||||
Name string
|
||||
Path string
|
||||
SizeBytes int64
|
||||
ModifiedAt time.Time
|
||||
ContentHash string
|
||||
}
|
||||
|
||||
// Source is the minimal capability Reconcile needs from a connector: list
|
||||
// every document in the synced drive/folder. m365connector.Client
|
||||
// implements this via a Graph "children" listing (see GraphSource).
|
||||
type Source interface {
|
||||
ListDocuments(ctx context.Context) ([]ExternalDocument, error)
|
||||
}
|
||||
|
||||
// Record is the stored reconciliation state for one synced document.
|
||||
type Record struct {
|
||||
ID int64
|
||||
Ref Ref
|
||||
ExternalID string
|
||||
Name string
|
||||
Path string
|
||||
SizeBytes int64
|
||||
ModifiedAt time.Time
|
||||
ExternalHash string
|
||||
NexarchHash string
|
||||
Conflict bool
|
||||
ConflictDetectedAt *time.Time
|
||||
}
|
||||
|
||||
// Result summarizes one Reconcile call.
|
||||
type Result struct {
|
||||
Created int
|
||||
Updated int
|
||||
Conflicts int
|
||||
}
|
||||
|
||||
// Store persists document reconciliation state in Postgres.
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Get returns the stored record for one external ID, or ErrNotFound.
|
||||
func (s *Store) Get(ctx context.Context, ref Ref, 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, external_id,
|
||||
name, path, size_bytes, modified_at, external_hash, nexarch_hash, conflict, conflict_detected_at
|
||||
FROM m365_doc_records
|
||||
WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND external_id = $4
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, externalID).Scan(
|
||||
&r.ID, &r.Ref.TenantSlug, &r.Ref.ConnectorName, &r.Ref.ConnectorInstanceID, &r.ExternalID,
|
||||
&r.Name, &r.Path, &r.SizeBytes, &r.ModifiedAt, &r.ExternalHash, &r.NexarchHash, &r.Conflict, &conflictAt,
|
||||
)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Record{}, ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return Record{}, fmt.Errorf("m365docsync: get: %w", err)
|
||||
}
|
||||
r.ConflictDetectedAt = conflictAt
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// ListByRef returns every stored record for a connector instance, for
|
||||
// inspection/testing.
|
||||
func (s *Store) ListByRef(ctx context.Context, ref Ref) ([]Record, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, tenant_slug, connector_name, connector_instance_id, external_id,
|
||||
name, path, size_bytes, modified_at, external_hash, nexarch_hash, conflict, conflict_detected_at
|
||||
FROM m365_doc_records
|
||||
WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
|
||||
ORDER BY external_id
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("m365docsync: 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.ExternalID,
|
||||
&r.Name, &r.Path, &r.SizeBytes, &r.ModifiedAt, &r.ExternalHash, &r.NexarchHash, &r.Conflict, &conflictAt); err != nil {
|
||||
return nil, fmt.Errorf("m365docsync: list: scan: %w", err)
|
||||
}
|
||||
r.ConflictDetectedAt = conflictAt
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// SetLocalHash simulates/records a NEXARCH-side edit to a synced document,
|
||||
// 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).
|
||||
func (s *Store) SetLocalHash(ctx context.Context, ref Ref, externalID string, nexarchHash string) error {
|
||||
tag, err := s.pool.Exec(ctx, `
|
||||
UPDATE m365_doc_records SET nexarch_hash = $5, updated_at = now()
|
||||
WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND external_id = $4
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, externalID, nexarchHash)
|
||||
if err != nil {
|
||||
return fmt.Errorf("m365docsync: set local hash: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResolveConflict clears a record's conflict flag and adopts resolvedHash
|
||||
// as the new NEXARCH-side state, so the next Reconcile call no longer
|
||||
// treats it as diverged. Never called automatically -- a human decides the
|
||||
// resolution.
|
||||
func (s *Store) ResolveConflict(ctx context.Context, ref Ref, externalID string, resolvedHash string) error {
|
||||
tag, err := s.pool.Exec(ctx, `
|
||||
UPDATE m365_doc_records
|
||||
SET nexarch_hash = $5, conflict = false, conflict_detected_at = NULL, updated_at = now()
|
||||
WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND external_id = $4
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, externalID, resolvedHash)
|
||||
if err != nil {
|
||||
return fmt.Errorf("m365docsync: resolve conflict: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Reconcile fetches every document from source and reconciles it against
|
||||
// stored state:
|
||||
// - unseen external_id -> inserted as a new NEXARCH document with its
|
||||
// metadata (AC1).
|
||||
// - external side changed, NEXARCH side unchanged since last sync ->
|
||||
// the external change (metadata + hash) is adopted.
|
||||
// - external side changed AND the NEXARCH side was independently edited
|
||||
// (via SetLocalHash) 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, source Source) (Result, error) {
|
||||
docs, err := source.ListDocuments(ctx)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("m365docsync: reconcile: list: %w", err)
|
||||
}
|
||||
|
||||
var res Result
|
||||
for _, doc := range docs {
|
||||
outcome, err := s.reconcileOne(ctx, ref, doc)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("m365docsync: reconcile %s: %w", doc.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, doc ExternalDocument) (outcome reconcileOutcome, err error) {
|
||||
var existingID int64
|
||||
var storedExternalHash, storedNexarchHash string
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
SELECT id, external_hash, nexarch_hash FROM m365_doc_records
|
||||
WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND external_id = $4
|
||||
FOR UPDATE
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, doc.ExternalID).
|
||||
Scan(&existingID, &storedExternalHash, &storedNexarchHash)
|
||||
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
_, insertErr := s.pool.Exec(ctx, `
|
||||
INSERT INTO m365_doc_records
|
||||
(tenant_slug, connector_name, connector_instance_id, external_id,
|
||||
name, path, size_bytes, modified_at, external_hash, nexarch_hash)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9)
|
||||
`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, doc.ExternalID,
|
||||
doc.Name, doc.Path, doc.SizeBytes, doc.ModifiedAt, doc.ContentHash)
|
||||
if insertErr != nil {
|
||||
return outcomeNoop, fmt.Errorf("insert: %w", insertErr)
|
||||
}
|
||||
return outcomeCreated, nil
|
||||
}
|
||||
if err != nil {
|
||||
return outcomeNoop, fmt.Errorf("lookup: %w", err)
|
||||
}
|
||||
|
||||
externalChanged := storedExternalHash != doc.ContentHash
|
||||
localChanged := storedNexarchHash != storedExternalHash
|
||||
if !externalChanged {
|
||||
return outcomeNoop, nil
|
||||
}
|
||||
if localChanged {
|
||||
_, err = s.pool.Exec(ctx, `
|
||||
UPDATE m365_doc_records
|
||||
SET name = $2, path = $3, size_bytes = $4, modified_at = $5, external_hash = $6,
|
||||
conflict = true, conflict_detected_at = now(), updated_at = now()
|
||||
WHERE id = $1
|
||||
`, existingID, doc.Name, doc.Path, doc.SizeBytes, doc.ModifiedAt, doc.ContentHash)
|
||||
if err != nil {
|
||||
return outcomeNoop, fmt.Errorf("mark conflict: %w", err)
|
||||
}
|
||||
return outcomeConflicted, nil
|
||||
}
|
||||
|
||||
_, err = s.pool.Exec(ctx, `
|
||||
UPDATE m365_doc_records
|
||||
SET name = $2, path = $3, size_bytes = $4, modified_at = $5, external_hash = $6, nexarch_hash = $6, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, existingID, doc.Name, doc.Path, doc.SizeBytes, doc.ModifiedAt, doc.ContentHash)
|
||||
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
|
||||
}
|
||||
|
||||
// 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 masterdatasync's Runner) --
|
||||
// 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("m365docsync: run due jobs: claim: %w", err)
|
||||
}
|
||||
n := 0
|
||||
for _, run := range runs {
|
||||
if run.SyncJobID != jobID {
|
||||
continue
|
||||
}
|
||||
_, reconcileErr := r.Store.Reconcile(ctx, ref, r.Source)
|
||||
if completeErr := r.Jobs.CompleteRun(ctx, run.ID, reconcileErr); completeErr != nil {
|
||||
return n, fmt.Errorf("m365docsync: run due jobs: complete run: %w", completeErr)
|
||||
}
|
||||
n++
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
@@ -0,0 +1,325 @@
|
||||
package m365docsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/connect/internal/credential"
|
||||
"gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate"
|
||||
"gitea.perlbach24.de/scripte/nexarch/connect/internal/syncjob"
|
||||
"gitea.perlbach24.de/scripte/nexarch/connect/internal/testdb"
|
||||
)
|
||||
|
||||
func newStore(t *testing.T) (*Store, *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_TENANT_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
pool := testdb.Pool(t, dsn)
|
||||
|
||||
root, err := filepath.Abs("../../migrations")
|
||||
if err != nil {
|
||||
t.Fatalf("migrationsverzeichnis: %v", err)
|
||||
}
|
||||
migrations, err := migrate.Load(root)
|
||||
if err != nil {
|
||||
t.Fatalf("migrationen laden: %v", err)
|
||||
}
|
||||
if _, err := migrate.Up(context.Background(), pool, migrations); err != nil {
|
||||
t.Fatalf("migrationen anwenden: %v", err)
|
||||
}
|
||||
return NewStore(pool), pool
|
||||
}
|
||||
|
||||
var testRef = Ref{TenantSlug: "acme", ConnectorName: "m365", ConnectorInstanceID: "inst-1"}
|
||||
|
||||
// docSource serves a fixed slice of documents, so a test can change what
|
||||
// "the drive" contains between Reconcile calls.
|
||||
type docSource struct{ docs []ExternalDocument }
|
||||
|
||||
func (s *docSource) ListDocuments(ctx context.Context) ([]ExternalDocument, error) {
|
||||
return s.docs, nil
|
||||
}
|
||||
|
||||
func mustTime(t *testing.T, s string) time.Time {
|
||||
t.Helper()
|
||||
ts, err := time.Parse(time.RFC3339, s)
|
||||
if err != nil {
|
||||
t.Fatalf("zeit parsen: %v", err)
|
||||
}
|
||||
return ts
|
||||
}
|
||||
|
||||
// AC1: ein Dokument, das nur auf der SharePoint/OneDrive-Seite existiert,
|
||||
// erscheint nach Sync in NEXARCH mit seinen Metadaten.
|
||||
func TestReconcile_NewExternalDocument_IsCreatedWithMetadata(t *testing.T) {
|
||||
store, _ := newStore(t)
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-1", Name: "Vertrag.pdf", Path: "/drive/root:/Vertraege", SizeBytes: 1024,
|
||||
ModifiedAt: mustTime(t, "2026-01-01T10:00:00Z"), ContentHash: "hash-a"},
|
||||
}}
|
||||
|
||||
res, err := store.Reconcile(context.Background(), testRef, src)
|
||||
if err != nil {
|
||||
t.Fatalf("reconcile: %v", err)
|
||||
}
|
||||
if res.Created != 1 || res.Updated != 0 || res.Conflicts != 0 {
|
||||
t.Fatalf("erwartet 1 created, 0 updated, 0 conflicts, bekam %+v", res)
|
||||
}
|
||||
|
||||
rec, err := store.Get(context.Background(), testRef, "doc-1")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if rec.Name != "Vertrag.pdf" || rec.Path != "/drive/root:/Vertraege" || rec.SizeBytes != 1024 {
|
||||
t.Fatalf("metadaten nicht korrekt uebernommen: %+v", rec)
|
||||
}
|
||||
if rec.Conflict {
|
||||
t.Fatalf("neues dokument darf nicht als konflikt markiert sein")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcile_NoChange_IsNoop(t *testing.T) {
|
||||
store, _ := newStore(t)
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-2", Name: "A.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-01T00:00:00Z"), ContentHash: "h1"},
|
||||
}}
|
||||
if _, err := store.Reconcile(context.Background(), testRef, src); err != nil {
|
||||
t.Fatalf("erster reconcile: %v", err)
|
||||
}
|
||||
res, err := store.Reconcile(context.Background(), testRef, src)
|
||||
if err != nil {
|
||||
t.Fatalf("zweiter reconcile: %v", err)
|
||||
}
|
||||
if res.Created != 0 || res.Updated != 0 || res.Conflicts != 0 {
|
||||
t.Fatalf("erwartet reines no-op, bekam %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcile_ExternalChangeOnly_IsAdopted(t *testing.T) {
|
||||
store, _ := newStore(t)
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-3", Name: "Alt.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-01T00:00:00Z"), ContentHash: "h1"},
|
||||
}}
|
||||
if _, err := store.Reconcile(context.Background(), testRef, src); err != nil {
|
||||
t.Fatalf("erster reconcile: %v", err)
|
||||
}
|
||||
|
||||
src.docs = []ExternalDocument{
|
||||
{ExternalID: "doc-3", Name: "Neu.pdf", Path: "/x", SizeBytes: 2, ModifiedAt: mustTime(t, "2026-01-02T00:00:00Z"), ContentHash: "h2"},
|
||||
}
|
||||
res, err := store.Reconcile(context.Background(), testRef, src)
|
||||
if err != nil {
|
||||
t.Fatalf("zweiter reconcile: %v", err)
|
||||
}
|
||||
if res.Updated != 1 || res.Conflicts != 0 {
|
||||
t.Fatalf("erwartet 1 updated, 0 conflicts, bekam %+v", res)
|
||||
}
|
||||
|
||||
rec, err := store.Get(context.Background(), testRef, "doc-3")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if rec.Name != "Neu.pdf" || rec.NexarchHash != "h2" {
|
||||
t.Fatalf("externe aenderung wurde nicht uebernommen: %+v", rec)
|
||||
}
|
||||
}
|
||||
|
||||
// AC2: gleichzeitige Aenderung auf beiden Seiten erzeugt einen erkennbaren
|
||||
// Konflikt statt eines stillen Datenverlusts.
|
||||
func TestReconcile_ConcurrentChangeBothSides_IsFlaggedNotOverwritten(t *testing.T) {
|
||||
store, _ := newStore(t)
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-4", Name: "Ausgangsstand.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-01T00:00:00Z"), ContentHash: "h1"},
|
||||
}}
|
||||
if _, err := store.Reconcile(context.Background(), testRef, src); err != nil {
|
||||
t.Fatalf("erster reconcile: %v", err)
|
||||
}
|
||||
|
||||
// NEXARCH-seitige Aenderung, unabhaengig vom naechsten Sync.
|
||||
if err := store.SetLocalHash(context.Background(), testRef, "doc-4", "h1-lokal-geaendert"); err != nil {
|
||||
t.Fatalf("set local hash: %v", err)
|
||||
}
|
||||
|
||||
// Gleichzeitig hat sich auch die externe Seite geaendert.
|
||||
src.docs = []ExternalDocument{
|
||||
{ExternalID: "doc-4", Name: "Extern-geaendert.pdf", Path: "/x", SizeBytes: 3, ModifiedAt: mustTime(t, "2026-01-03T00:00:00Z"), ContentHash: "h3"},
|
||||
}
|
||||
|
||||
res, err := store.Reconcile(context.Background(), testRef, src)
|
||||
if err != nil {
|
||||
t.Fatalf("reconcile mit konflikt: %v", err)
|
||||
}
|
||||
if res.Conflicts != 1 || res.Updated != 0 {
|
||||
t.Fatalf("erwartet 1 conflict, 0 updated, bekam %+v", res)
|
||||
}
|
||||
|
||||
rec, err := store.Get(context.Background(), testRef, "doc-4")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if !rec.Conflict {
|
||||
t.Fatalf("record haette als conflict markiert sein muessen")
|
||||
}
|
||||
if rec.NexarchHash != "h1-lokal-geaendert" {
|
||||
t.Fatalf("nexarch_hash wurde stillschweigend ueberschrieben: %s", rec.NexarchHash)
|
||||
}
|
||||
if rec.ExternalHash != "h3" {
|
||||
t.Fatalf("external_hash haette den neuen externen stand tracken sollen: %s", rec.ExternalHash)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveConflict_ClearsFlagAndAdoptsResolvedHash(t *testing.T) {
|
||||
store, _ := newStore(t)
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-5", Name: "A.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-01T00:00:00Z"), ContentHash: "h1"},
|
||||
}}
|
||||
if _, err := store.Reconcile(context.Background(), testRef, src); err != nil {
|
||||
t.Fatalf("erster reconcile: %v", err)
|
||||
}
|
||||
if err := store.SetLocalHash(context.Background(), testRef, "doc-5", "h-lokal"); err != nil {
|
||||
t.Fatalf("set local: %v", err)
|
||||
}
|
||||
src.docs = []ExternalDocument{
|
||||
{ExternalID: "doc-5", Name: "A.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-02T00:00:00Z"), ContentHash: "h-extern"},
|
||||
}
|
||||
if _, err := store.Reconcile(context.Background(), testRef, src); err != nil {
|
||||
t.Fatalf("konflikt-reconcile: %v", err)
|
||||
}
|
||||
|
||||
if err := store.ResolveConflict(context.Background(), testRef, "doc-5", "h-aufgeloest"); err != nil {
|
||||
t.Fatalf("resolve conflict: %v", err)
|
||||
}
|
||||
rec, err := store.Get(context.Background(), testRef, "doc-5")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if rec.Conflict {
|
||||
t.Fatalf("conflict-flag haette geloescht sein sollen")
|
||||
}
|
||||
if rec.NexarchHash != "h-aufgeloest" {
|
||||
t.Fatalf("nexarch_hash = %s, erwartet aufgeloesten stand", rec.NexarchHash)
|
||||
}
|
||||
|
||||
// Ein weiterer Reconcile-Lauf mit unveraendertem externen Stand darf
|
||||
// den aufgeloesten Konflikt nicht wieder aufreissen.
|
||||
res, err := store.Reconcile(context.Background(), testRef, src)
|
||||
if err != nil {
|
||||
t.Fatalf("reconcile nach resolve: %v", err)
|
||||
}
|
||||
if res.Conflicts != 0 {
|
||||
t.Fatalf("erwartet keinen erneuten conflict nach resolve, bekam %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
// AC3: der Abgleich laeuft nach Zeitplan ueber die Sync-Job-Verwaltung
|
||||
// (ENG-03) statt eines eigenen Cron-Mechanismus.
|
||||
func TestRunner_RunDueJobs_ExecutesReconcileOnSchedule(t *testing.T) {
|
||||
store, pool := newStore(t)
|
||||
jobs := syncjob.NewStore(pool)
|
||||
|
||||
jobID, err := jobs.Create(context.Background(), syncjob.Ref{
|
||||
TenantSlug: testRef.TenantSlug, ConnectorName: testRef.ConnectorName, ConnectorInstanceID: testRef.ConnectorInstanceID,
|
||||
}, 60)
|
||||
if err != nil {
|
||||
t.Fatalf("create job: %v", err)
|
||||
}
|
||||
|
||||
src := &docSource{docs: []ExternalDocument{
|
||||
{ExternalID: "doc-6", Name: "Zeitplan.pdf", Path: "/x", SizeBytes: 1, ModifiedAt: mustTime(t, "2026-01-01T00:00:00Z"), ContentHash: "h1"},
|
||||
}}
|
||||
runner := &Runner{Jobs: jobs, Store: store, Source: src}
|
||||
|
||||
n, err := runner.RunDueJobs(context.Background(), testRef, jobID, time.Now().Add(time.Second))
|
||||
if err != nil {
|
||||
t.Fatalf("run due jobs: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("erwartet 1 ausgefuehrten job, bekam %d", n)
|
||||
}
|
||||
|
||||
rec, err := store.Get(context.Background(), testRef, "doc-6")
|
||||
if err != nil {
|
||||
t.Fatalf("get nach scheduled run: %v", err)
|
||||
}
|
||||
if rec.Name != "Zeitplan.pdf" {
|
||||
t.Fatalf("scheduled reconcile hat das dokument nicht angelegt: %+v", rec)
|
||||
}
|
||||
|
||||
runs, err := jobs.ListRuns(context.Background(), jobID)
|
||||
if err != nil {
|
||||
t.Fatalf("list runs: %v", err)
|
||||
}
|
||||
if len(runs) != 1 || runs[0].Status != syncjob.RunStatusSucceeded {
|
||||
t.Fatalf("erwartet genau 1 erfolgreichen run, bekam %+v", runs)
|
||||
}
|
||||
}
|
||||
|
||||
// AC1 end-to-end gegen einen gefakten Graph-Server (kein echter M365-
|
||||
// Tenant verfuegbar, daher wie schon in M365-01 dokumentiert ein
|
||||
// httptest-Mock statt eines Tests gegen echtes SharePoint/OneDrive).
|
||||
func TestGraphSource_ListDocuments_ParsesChildrenListing(t *testing.T) {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/v1.0/drives/drive-1/root/children", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Write([]byte(`{"value":[
|
||||
{"id":"item-1","name":"Bericht.docx","size":2048,"lastModifiedDateTime":"2026-01-05T12:00:00Z",
|
||||
"parentReference":{"path":"/drives/drive-1/root:/Berichte"},
|
||||
"file":{"hashes":{"quickXorHash":"abc123"}}},
|
||||
{"id":"folder-1","name":"Unterordner","folder":{}}
|
||||
]}`))
|
||||
})
|
||||
srv := httptest.NewServer(mux)
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
reader := &httpGraphReader{baseURL: srv.URL + "/v1.0"}
|
||||
src := GraphSource{Graph: reader, DriveID: "drive-1"}
|
||||
|
||||
docs, err := src.ListDocuments(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("list documents: %v", err)
|
||||
}
|
||||
if len(docs) != 1 {
|
||||
t.Fatalf("erwartet 1 dokument (ordner uebersprungen), bekam %d: %+v", len(docs), docs)
|
||||
}
|
||||
doc := docs[0]
|
||||
if doc.ExternalID != "item-1" || doc.Name != "Bericht.docx" || doc.SizeBytes != 2048 || doc.ContentHash != "abc123" {
|
||||
t.Fatalf("dokument falsch geparst: %+v", doc)
|
||||
}
|
||||
if doc.Path != "/drives/drive-1/root:/Berichte" {
|
||||
t.Fatalf("pfad falsch geparst: %s", doc.Path)
|
||||
}
|
||||
}
|
||||
|
||||
// httpGraphReader ruft testweise direkt einen echten HTTP-Endpunkt auf, im
|
||||
// selben Format wie m365connector.Client.Read (GET <base>+path).
|
||||
type httpGraphReader struct{ baseURL string }
|
||||
|
||||
func (r *httpGraphReader) Read(ctx context.Context, secret credential.Secret, path string) ([]byte, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.baseURL+path, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var buf []byte
|
||||
dec := json.NewDecoder(resp.Body)
|
||||
var raw json.RawMessage
|
||||
if err := dec.Decode(&raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
buf = raw
|
||||
return buf, nil
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS m365_doc_records;
|
||||
@@ -0,0 +1,26 @@
|
||||
-- M365-02: SharePoint/OneDrive-Dokumentabgleich. Getrennte Hash-Staende fuer
|
||||
-- die zuletzt von Graph gesehene Fassung und die NEXARCH-seitig zuletzt
|
||||
-- uebernommene Fassung, analog zu master_data_records (ERP-03) -- ein
|
||||
-- Auseinanderlaufen beider Hashes bei gleichzeitiger externer Aenderung ist
|
||||
-- ein Konflikt (AC2) statt einer stillen Ueberschreibung.
|
||||
CREATE TABLE m365_doc_records (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
tenant_slug TEXT NOT NULL,
|
||||
connector_name TEXT NOT NULL,
|
||||
connector_instance_id TEXT NOT NULL,
|
||||
external_id TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
path TEXT NOT NULL,
|
||||
size_bytes BIGINT NOT NULL,
|
||||
modified_at TIMESTAMPTZ NOT NULL,
|
||||
external_hash TEXT NOT NULL,
|
||||
nexarch_hash TEXT NOT NULL,
|
||||
conflict BOOLEAN NOT NULL DEFAULT false,
|
||||
conflict_detected_at TIMESTAMPTZ,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
UNIQUE (tenant_slug, connector_name, connector_instance_id, external_id)
|
||||
);
|
||||
|
||||
CREATE INDEX m365_doc_records_ref_idx
|
||||
ON m365_doc_records (tenant_slug, connector_name, connector_instance_id);
|
||||
Regular → Executable
+1
@@ -21,6 +21,7 @@ SECURITY_PACKAGES=(
|
||||
"internal/dms2importapi"
|
||||
"internal/fieldmapping"
|
||||
"internal/masterdatasync"
|
||||
"internal/m365docsync"
|
||||
"internal/syncjob"
|
||||
"internal/statusapi"
|
||||
)
|
||||
|
||||
Reference in new issue
Block a user