From f1a8d141335dd67e35a8341c8ec39750467c4292 Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 10:35:00 +0100 Subject: [PATCH 1/6] feat(mapping): add sync command Adds `fga mapping sync`, which reads JSONL from stdin (or --input), evaluates each record against a mapping file, and applies the resulting tuple writes and deletes to a store. Key behaviours: - Progress is reported to stderr as a single updating line (TTY) or throttled once-per-second lines (non-TTY). - --dry-run reads the store and prints planned operations to stdout (suppressed at an interactive terminal). - --continue-on-error skips failed records and exits non-zero. - --quiet suppresses progress; errors are always shown. - Deletes are issued before writes so condition changes land correctly. - Duplicate per-chunk SDK errors are deduplicated before surfacing. Includes unit tests for syncMapping and the client wrappers, and commander integration tests covering the main flag combinations. --- cmd/mapping/client.go | 215 ++++++++++++++++ cmd/mapping/client_test.go | 205 ++++++++++++++++ cmd/mapping/mapping.go | 1 + cmd/mapping/sync.go | 380 +++++++++++++++++++++++++++++ cmd/mapping/sync_test.go | 202 +++++++++++++++ go.mod | 4 +- go.sum | 4 +- tests/fixtures/sync-event.json | 1 + tests/fixtures/sync-mapping.yaml | 7 + tests/mapping-sync-test-cases.yaml | 54 ++++ 10 files changed, 1069 insertions(+), 4 deletions(-) create mode 100644 cmd/mapping/client.go create mode 100644 cmd/mapping/client_test.go create mode 100644 cmd/mapping/sync.go create mode 100644 cmd/mapping/sync_test.go create mode 100644 tests/fixtures/sync-event.json create mode 100644 tests/fixtures/sync-mapping.yaml create mode 100644 tests/mapping-sync-test-cases.yaml diff --git a/cmd/mapping/client.go b/cmd/mapping/client.go new file mode 100644 index 00000000..321e5153 --- /dev/null +++ b/cmd/mapping/client.go @@ -0,0 +1,215 @@ +package mapping + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + + openfga "github.com/openfga/go-sdk" + sdkclient "github.com/openfga/go-sdk/client" + "github.com/openfga/mapper/apply" + "github.com/openfga/mapper/language" + + internaltuple "github.com/openfga/cli/internal/tuple" +) + +const maxWriteChunkSize = 40 + +// fgaTupleClient adapts the OpenFGA SDK client to the apply.TupleClient interface. +type fgaTupleClient struct { + inner sdkclient.SdkClient +} + +func (c *fgaTupleClient) ReadTuples(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) { + body := &sdkclient.ClientReadRequest{} + if filter.User != "" { + body.User = &filter.User + } + if filter.Relation != "" { + body.Relation = &filter.Relation + } + if filter.Object != "" { + body.Object = &filter.Object + } + + resp, err := internaltuple.Read(ctx, c.inner, body, 0, internaltuple.DefaultReadPageSize, nil) + if err != nil { + return nil, err + } + + tuples := make([]language.Tuple, 0, len(resp.Tuples)) + for _, t := range resp.Tuples { + key := t.GetKey() + tuple := language.Tuple{ + User: key.GetUser(), + Relation: key.GetRelation(), + Object: key.GetObject(), + } + if cond, ok := key.GetConditionOk(); ok { + tuple.Condition = cond.GetName() + tuple.Context = cond.GetContext() + } + tuples = append(tuples, tuple) + } + + return tuples, nil +} + +func (c *fgaTupleClient) WriteTuples(ctx context.Context, tuples []language.Tuple) error { + writes, deletes := partitionSyncTuples(tuples) + if len(writes) == 0 && len(deletes) == 0 { + return nil + } + + writeOpts := sdkclient.ClientWriteOptions{ + Conflict: sdkclient.ClientWriteConflictOptions{ + OnDuplicateWrites: sdkclient.CLIENT_WRITE_REQUEST_ON_DUPLICATE_WRITES_IGNORE, + OnMissingDeletes: sdkclient.CLIENT_WRITE_REQUEST_ON_MISSING_DELETES_IGNORE, + }, + Transaction: &sdkclient.TransactionOptions{ + Disable: true, + MaxPerChunk: maxWriteChunkSize, + }, + } + + // Deletes must be issued before writes. The OpenFGA server rejects a write to a + // URO that already holds a tuple with a different condition (TupleConditionConflictError), + // so a condition change (delete old + write new on same URO) requires the delete to + // land first. The SDK with Transaction.Disable sends writes first then deletes, so we + // split them into two calls. + if len(deletes) > 0 { + resp, err := c.inner.Write(ctx).Options(writeOpts).Body(sdkclient.ClientWriteRequest{ + Deletes: deletes, + }).Execute() + if err != nil { + return fmt.Errorf("deleting tuples: %w", err) + } + var errs []error + seen := make(map[string]struct{}) + for _, d := range resp.Deletes { + if d.Status == sdkclient.FAILURE { + var e error + if d.Error != nil { + e = d.Error + } else { + e = fmt.Errorf("delete failed: %s %s %s", d.TupleKey.User, d.TupleKey.Relation, d.TupleKey.Object) + } + if _, ok := seen[e.Error()]; !ok { + seen[e.Error()] = struct{}{} + errs = append(errs, e) + } + } + } + if len(errs) > 0 { + return fmt.Errorf("deleting tuples: %w", errors.Join(errs...)) + } + } + + if len(writes) > 0 { + resp, err := c.inner.Write(ctx).Options(writeOpts).Body(sdkclient.ClientWriteRequest{ + Writes: writes, + }).Execute() + if err != nil { + return fmt.Errorf("writing tuples: %w", err) + } + var errs []error + seen := make(map[string]struct{}) + for _, w := range resp.Writes { + if w.Status == sdkclient.FAILURE { + var e error + if w.Error != nil { + e = w.Error + } else { + e = fmt.Errorf("write failed: %s %s %s", w.TupleKey.User, w.TupleKey.Relation, w.TupleKey.Object) + } + if _, ok := seen[e.Error()]; !ok { + seen[e.Error()] = struct{}{} + errs = append(errs, e) + } + } + } + if len(errs) > 0 { + return fmt.Errorf("writing tuples: %w", errors.Join(errs...)) + } + } + + return nil +} + +// countingClient wraps a TupleClient and tracks how many tuples were written/deleted. +// Reset writes and dels between records. Correctness depends on apply setting t.Action +// on every tuple it passes to WriteTuples; an unset Action is treated as a write. +type countingClient struct { + inner apply.TupleClient + writes int + dels int +} + +func (c *countingClient) ReadTuples(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) { + return c.inner.ReadTuples(ctx, filter) +} + +func (c *countingClient) WriteTuples(ctx context.Context, tuples []language.Tuple) error { + for _, t := range tuples { + if t.Action == language.ActionDelete { + c.dels++ + } else { + c.writes++ + } + } + return c.inner.WriteTuples(ctx, tuples) +} + +// printOnlyClient reads from inner but prints planned writes/deletes to out instead of applying them. +// Used for --dry-run. +type printOnlyClient struct { + inner apply.TupleClient + out io.Writer +} + +func (c *printOnlyClient) ReadTuples(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) { + return c.inner.ReadTuples(ctx, filter) +} + +func (c *printOnlyClient) WriteTuples(_ context.Context, tuples []language.Tuple) error { + enc := json.NewEncoder(c.out) + for _, t := range tuples { + if err := enc.Encode(toOpTuple(t)); err != nil { + return fmt.Errorf("encoding dry-run output: %w", err) + } + } + return nil +} + +func partitionSyncTuples(tuples []language.Tuple) ([]sdkclient.ClientTupleKey, []sdkclient.ClientTupleKeyWithoutCondition) { + var writes []sdkclient.ClientTupleKey + var deletes []sdkclient.ClientTupleKeyWithoutCondition + + for _, t := range tuples { + if t.Action == language.ActionDelete { + deletes = append(deletes, sdkclient.ClientTupleKeyWithoutCondition{ + User: t.User, + Relation: t.Relation, + Object: t.Object, + }) + } else { + key := sdkclient.ClientTupleKey{ + User: t.User, + Relation: t.Relation, + Object: t.Object, + } + if t.Condition != "" { + rc := &openfga.RelationshipCondition{Name: t.Condition} + if len(t.Context) > 0 { + rc.Context = &t.Context + } + key.Condition = rc + } + writes = append(writes, key) + } + } + + return writes, deletes +} diff --git a/cmd/mapping/client_test.go b/cmd/mapping/client_test.go new file mode 100644 index 00000000..491c2a09 --- /dev/null +++ b/cmd/mapping/client_test.go @@ -0,0 +1,205 @@ +package mapping + +import ( + "bytes" + "context" + "errors" + "strings" + "testing" + + "github.com/openfga/mapper/language" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type alwaysErrWriter struct{} + +func (e *alwaysErrWriter) Write(_ []byte) (int, error) { + return 0, errors.New("write error") +} + +func TestCountingClient(t *testing.T) { + t.Parallel() + + t.Run("counts writes and deletes", func(t *testing.T) { + t.Parallel() + + cc := &countingClient{inner: &mockTupleClient{}} + tuples := []language.Tuple{ + {User: "user:anne", Relation: "member", Object: "org:acme", Action: language.ActionWrite}, + {User: "user:bob", Relation: "member", Object: "org:acme", Action: language.ActionWrite}, + {User: "user:eve", Relation: "viewer", Object: "org:acme", Action: language.ActionDelete}, + } + + require.NoError(t, cc.WriteTuples(context.Background(), tuples)) + assert.Equal(t, 2, cc.writes) + assert.Equal(t, 1, cc.dels) + }) + + t.Run("delegates WriteTuples to inner", func(t *testing.T) { + t.Parallel() + + var received []language.Tuple + inner := &mockTupleClient{ + writeFn: func(_ context.Context, tuples []language.Tuple) error { + received = tuples + return nil + }, + } + cc := &countingClient{inner: inner} + tuples := []language.Tuple{ + {User: "user:anne", Relation: "member", Object: "org:acme"}, + } + + require.NoError(t, cc.WriteTuples(context.Background(), tuples)) + assert.Equal(t, tuples, received) + }) + + t.Run("delegates ReadTuples to inner", func(t *testing.T) { + t.Parallel() + + expected := []language.Tuple{{User: "user:anne", Relation: "member", Object: "org:acme"}} + inner := &mockTupleClient{ + readFn: func(_ context.Context, _ language.TupleFilter) ([]language.Tuple, error) { + return expected, nil + }, + } + cc := &countingClient{inner: inner} + + got, err := cc.ReadTuples(context.Background(), language.TupleFilter{}) + require.NoError(t, err) + assert.Equal(t, expected, got) + }) +} + +func TestPrintOnlyClient(t *testing.T) { + t.Parallel() + + t.Run("WriteTuples encodes tuples as JSONL with op field", func(t *testing.T) { + t.Parallel() + + var out bytes.Buffer + pc := &printOnlyClient{inner: &mockTupleClient{}, out: &out} + tuples := []language.Tuple{ + {User: "user:anne", Relation: "member", Object: "org:acme", Action: language.ActionWrite}, + {User: "user:bob", Relation: "viewer", Object: "org:acme", Action: language.ActionDelete}, + } + + require.NoError(t, pc.WriteTuples(context.Background(), tuples)) + + lines := strings.Split(strings.TrimSpace(out.String()), "\n") + require.Len(t, lines, 2) + assert.Contains(t, lines[0], `"user:anne"`) + assert.Contains(t, lines[0], `"op"`) + assert.Contains(t, lines[1], `"user:bob"`) + }) + + t.Run("WriteTuples returns error when writer fails", func(t *testing.T) { + t.Parallel() + + pc := &printOnlyClient{inner: &mockTupleClient{}, out: &alwaysErrWriter{}} + tuples := []language.Tuple{ + {User: "user:anne", Relation: "member", Object: "org:acme"}, + } + + err := pc.WriteTuples(context.Background(), tuples) + require.Error(t, err) + assert.Contains(t, err.Error(), "encoding dry-run output") + }) + + t.Run("delegates ReadTuples to inner", func(t *testing.T) { + t.Parallel() + + expected := []language.Tuple{{User: "user:anne", Relation: "member", Object: "org:acme"}} + inner := &mockTupleClient{ + readFn: func(_ context.Context, _ language.TupleFilter) ([]language.Tuple, error) { + return expected, nil + }, + } + pc := &printOnlyClient{inner: inner, out: &bytes.Buffer{}} + + got, err := pc.ReadTuples(context.Background(), language.TupleFilter{}) + require.NoError(t, err) + assert.Equal(t, expected, got) + }) +} + +func TestPartitionSyncTuples(t *testing.T) { + t.Parallel() + + t.Run("write tuple goes to writes slice", func(t *testing.T) { + t.Parallel() + + tuple := language.Tuple{ + User: "user:anne", + Relation: "member", + Object: "org:acme", + Action: language.ActionWrite, + } + writes, deletes := partitionSyncTuples([]language.Tuple{tuple}) + require.Len(t, writes, 1) + assert.Empty(t, deletes) + assert.Equal(t, "user:anne", writes[0].User) + assert.Equal(t, "member", writes[0].Relation) + assert.Equal(t, "org:acme", writes[0].Object) + }) + + t.Run("delete tuple goes to deletes slice", func(t *testing.T) { + t.Parallel() + + tuple := language.Tuple{ + User: "user:anne", + Relation: "member", + Object: "org:acme", + Action: language.ActionDelete, + } + writes, deletes := partitionSyncTuples([]language.Tuple{tuple}) + assert.Empty(t, writes) + require.Len(t, deletes, 1) + assert.Equal(t, "user:anne", deletes[0].User) + assert.Equal(t, "member", deletes[0].Relation) + assert.Equal(t, "org:acme", deletes[0].Object) + }) + + t.Run("write tuple with condition sets Condition on write key", func(t *testing.T) { + t.Parallel() + + tuple := language.Tuple{ + User: "user:anne", + Relation: "owner", + Object: "group:eng", + Condition: "check_ip", + } + writes, _ := partitionSyncTuples([]language.Tuple{tuple}) + require.Len(t, writes, 1) + require.NotNil(t, writes[0].Condition) + assert.Equal(t, "check_ip", writes[0].Condition.Name) + assert.Nil(t, writes[0].Condition.Context) + }) + + t.Run("write tuple with condition and context sets Context on write key", func(t *testing.T) { + t.Parallel() + + ctx := map[string]any{"ip_addr": "192.168.1.1"} + tuple := language.Tuple{ + User: "user:anne", + Relation: "owner", + Object: "group:eng", + Condition: "check_ip", + Context: ctx, + } + writes, _ := partitionSyncTuples([]language.Tuple{tuple}) + require.Len(t, writes, 1) + require.NotNil(t, writes[0].Condition) + require.NotNil(t, writes[0].Condition.Context) + assert.Equal(t, ctx, *writes[0].Condition.Context) + }) + + t.Run("empty input returns nil slices", func(t *testing.T) { + t.Parallel() + + writes, deletes := partitionSyncTuples(nil) + assert.Nil(t, writes) + assert.Nil(t, deletes) + }) +} diff --git a/cmd/mapping/mapping.go b/cmd/mapping/mapping.go index 2637ec35..ddc0a1fc 100644 --- a/cmd/mapping/mapping.go +++ b/cmd/mapping/mapping.go @@ -75,4 +75,5 @@ func init() { MappingCmd.AddCommand(testCmd) MappingCmd.AddCommand(initCmd) MappingCmd.AddCommand(evaluateCmd) + MappingCmd.AddCommand(syncCmd) } diff --git a/cmd/mapping/sync.go b/cmd/mapping/sync.go new file mode 100644 index 00000000..926838cd --- /dev/null +++ b/cmd/mapping/sync.go @@ -0,0 +1,380 @@ +package mapping + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "strings" + "time" + + "github.com/charmbracelet/lipgloss" + "github.com/mattn/go-isatty" + "github.com/openfga/mapper" + "github.com/openfga/mapper/apply" + "github.com/spf13/cobra" + + "github.com/openfga/cli/internal/cmdutils" +) + +var errSyncRecordsFailed = errors.New("one or more records failed") + +// errStyles holds lipgloss styles bound to a specific output writer. +type errStyles struct { + time lipgloss.Style + writes lipgloss.Style + deletes lipgloss.Style + dim lipgloss.Style + ok lipgloss.Style + errStyle lipgloss.Style +} + +func newErrStyles(w io.Writer) errStyles { + r := lipgloss.NewRenderer(w) + return errStyles{ + time: r.NewStyle().Faint(true), + writes: r.NewStyle().Foreground(lipgloss.Color("10")).Bold(true), + deletes: r.NewStyle().Foreground(lipgloss.Color("9")).Bold(true), + dim: r.NewStyle().Faint(true), + ok: r.NewStyle().Foreground(lipgloss.Color("10")), + errStyle: r.NewStyle().Foreground(lipgloss.Color("9")), + } +} + +var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} + +type syncOptions struct { + quiet bool + dryRun bool + continueOnError bool +} + +type syncStats struct { + records int + writes int + deletes int + errors int + frame int +} + +type statusPrinter struct { + out io.Writer + isTTY bool + quiet bool + styles errStyles + lastRender time.Time +} + +func newStatusPrinter(out io.Writer, quiet bool) *statusPrinter { + isTTY := false + if f, ok := out.(*os.File); ok { + isTTY = isatty.IsTerminal(f.Fd()) + } + return &statusPrinter{out: out, isTTY: isTTY, quiet: quiet, styles: newErrStyles(out)} +} + +func (p *statusPrinter) render(stats syncStats, suffix string) { + if p.quiet { + return + } + + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + spin := p.styles.dim.Render(spinnerFrames[stats.frame%len(spinnerFrames)]) + w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) + d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) + count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) + + line := fmt.Sprintf("%s %s %s %s %s", ts, spin, w, d, count) + if suffix != "" { + line += " " + suffix + } + + if p.isTTY { + fmt.Fprintf(p.out, "\r%s\x1b[K", line) + } else if time.Since(p.lastRender) >= time.Second { + p.lastRender = time.Now() + fmt.Fprintln(p.out, line) + } +} + +func (p *statusPrinter) done(stats syncStats) { + if p.quiet { + return + } + + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + tick := p.styles.ok.Render("✓") + w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) + d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) + count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) + + line := fmt.Sprintf("%s %s %s %s %s", ts, tick, w, d, count) + if stats.errors > 0 { + line += " " + p.styles.errStyle.Render(fmt.Sprintf("%d errors", stats.errors)) + } + + if p.isTTY { + fmt.Fprintf(p.out, "\r%s\x1b[K\n", line) + } else { + fmt.Fprintln(p.out, line) + } +} + +func (p *statusPrinter) errLine(msg string) { + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + x := p.styles.errStyle.Render("✗") + if p.isTTY { + fmt.Fprintf(p.out, "\r%s %s %s\x1b[K\n", ts, x, msg) + } else { + fmt.Fprintf(p.out, "%s %s %s\n", ts, x, msg) + } +} + +func syncMapping( + ctx context.Context, + path string, + opts syncOptions, + base apply.TupleClient, + inputReader io.Reader, + out, errOut io.Writer, +) error { + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("reading %s: %w", path, err) + } + + compiled, compileErr := mapper.Compile(data) + if compileErr != nil { + fmt.Fprintln(errOut, mapper.DiagnosticsFrom(compileErr)) + return errMappingInvalid + } + + // For dry-run, wrap the base client so writes are printed instead of applied. + // Only emit JSONL when stdout is redirected — at a TTY it's unreadable at scale. + var writeTarget apply.TupleClient = base + if opts.dryRun { + dryOut := io.Writer(io.Discard) + if f, ok := out.(*os.File); ok && !isatty.IsTerminal(f.Fd()) { + dryOut = out + } + writeTarget = &printOnlyClient{inner: base, out: dryOut} + } + + // countingClient is shared across records; reset before each call. + cc := &countingClient{inner: writeTarget} + rec := apply.New(cc) + + printer := newStatusPrinter(errOut, opts.quiet) + stats := syncStats{} + + heartbeat := time.NewTicker(30 * time.Second) + defer heartbeat.Stop() + + type recordResult struct { + writes int + deletes int + err error + } + + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + + // The goroutine lets the main select loop remain responsive during blocking reads. + // ReadString is not context-cancellable — on early return the goroutine is process-scoped + // and exits after the next line arrives or the input fd closes. + results := make(chan recordResult) + go func() { + defer close(results) + reader := bufio.NewReader(inputReader) + for { + line, readErr := reader.ReadString('\n') + if line != "" { + trimmed := strings.TrimSpace(line) + if trimmed != "" { + w, d, recErr := processSyncRecord(runCtx, compiled, rec, cc, trimmed) + select { + case results <- recordResult{writes: w, deletes: d, err: recErr}: + case <-runCtx.Done(): + return + } + } + } + if readErr != nil { + if !errors.Is(readErr, io.EOF) { + select { + case results <- recordResult{err: fmt.Errorf("reading input: %w", readErr)}: + case <-runCtx.Done(): + } + } + return + } + } + }() + + hadFailure := false + for { + select { + case res, ok := <-results: + if !ok { + printer.done(stats) + if hadFailure { + return errSyncRecordsFailed + } + return nil + } + if res.err != nil { + // --continue-on-error skips the whole record on any error (validation, + // read, write). Each record is independent so this is safe; the ✗ line + // on stderr makes skipped records visible. + if !opts.continueOnError { + cancel() + printer.errLine(res.err.Error()) + return res.err + } + hadFailure = true + stats.errors++ + stats.records++ + stats.frame++ + printer.errLine(res.err.Error()) + } else { + stats.records++ + stats.writes += res.writes + stats.deletes += res.deletes + stats.frame++ + printer.render(stats, "") + } + + case <-heartbeat.C: + stats.frame++ + printer.render(stats, printer.styles.dim.Render("waiting...")) + + case <-ctx.Done(): + cancel() + printer.errLine(ctx.Err().Error()) + return ctx.Err() + } + } +} + +func processSyncRecord( + ctx context.Context, + compiled *mapper.Mapping, + rec apply.Executor, + cc *countingClient, + line string, +) (writes, deletes int, err error) { + var event map[string]any + if err := json.Unmarshal([]byte(line), &event); err != nil { + return 0, 0, fmt.Errorf("reading input JSON: %w", err) + } + if event == nil { + return 0, 0, fmt.Errorf("reading input JSON: %w", errNonObjectRecord) + } + + result, err := compiled.Evaluate(ctx, event) + if err != nil { + return 0, 0, fmt.Errorf("evaluating mapping: %w", err) + } + + cc.writes = 0 + cc.dels = 0 + + if err := rec.Execute(ctx, result); err != nil { + return 0, 0, err + } + + return cc.writes, cc.dels, nil +} + + +var ( + syncInputFile string + syncDryRun bool + syncContinueOnError bool + syncQuiet bool +) + +var syncCmd = &cobra.Command{ + Use: "sync [mapping-file]", + Short: "Apply a mapping against JSON input, writing results to a store", + Long: `Reads JSONL from stdin (or --input) and evaluates it against the mapping file, +then applies the resulting tuple writes and deletes to the store. + +Rules using tuple_filters read current store state, diff against desired state, +and derive concrete writes and deletes before applying them. + +Progress is reported to stderr as a single updating line showing running write +and delete counts. On a non-interactive terminal each update is a new line. +Use --quiet to suppress all progress output (errors are always shown). + +--dry-run evaluates and expands filters (reading the store) but prints the +planned operations to stdout instead of writing them. +--continue-on-error skips records that fail to evaluate or apply (shown on +stderr) and exits non-zero if any were skipped.`, + Example: ` echo '{"id":"anne","org":"acme"}' | fga mapping sync mapping.yaml --store-id $STORE_ID + fga mapping sync mapping.yaml --input events.jsonl --store-id $STORE_ID --dry-run + fga mapping sync mapping.yaml --store-id $STORE_ID --continue-on-error < events.jsonl`, + Args: cobra.RangeArgs(0, 1), + RunE: func(cmd *cobra.Command, args []string) error { + clientConfig := cmdutils.GetClientConfig(cmd) + errStream := cmd.ErrOrStderr() + + fgaSDKClient, err := clientConfig.GetFgaClient() + if err != nil { + return fmt.Errorf("configuring FGA client: %w", err) + } + + path := promptMappingFile(args, errStream) + + inputReader := cmd.InOrStdin() + if syncInputFile != "" { + file, err := os.Open(syncInputFile) + if err != nil { + return fmt.Errorf("opening input %s: %w", syncInputFile, err) + } + defer file.Close() + inputReader = file + } + + opts := syncOptions{ + quiet: syncQuiet, + dryRun: syncDryRun, + continueOnError: syncContinueOnError, + } + + tupleClient := &fgaTupleClient{inner: fgaSDKClient} + + err = syncMapping( + cmd.Context(), path, opts, + tupleClient, inputReader, + cmd.OutOrStdout(), errStream, + ) + if errors.Is(err, errMappingInvalid) { + os.Exit(2) + } + + return err + }, +} + +func init() { + syncCmd.Flags().String("store-id", "", "Store ID") + syncCmd.Flags().String("model-id", "", "Authorization Model ID") + if err := syncCmd.MarkFlagRequired("store-id"); err != nil { + panic(err) + } + syncCmd.Flags().StringVar( + &syncInputFile, "input", "", "Path to JSONL input file, one JSON object per line (default: stdin)") + syncCmd.Flags().BoolVar( + &syncDryRun, "dry-run", false, + "Read the store and print planned operations to stdout without writing (suppressed at an interactive terminal)") + syncCmd.Flags().BoolVar( + &syncContinueOnError, "continue-on-error", false, + "Skip records that fail to evaluate or apply (shown on stderr) and exit non-zero if any were skipped") + syncCmd.Flags().BoolVar( + &syncQuiet, "quiet", false, + "Suppress progress output; only errors are written to stderr") +} diff --git a/cmd/mapping/sync_test.go b/cmd/mapping/sync_test.go new file mode 100644 index 00000000..db9af698 --- /dev/null +++ b/cmd/mapping/sync_test.go @@ -0,0 +1,202 @@ +package mapping + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "strings" + "testing" + + "github.com/openfga/mapper/apply" + "github.com/openfga/mapper/language" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type mockTupleClient struct { + readFn func(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) + writeFn func(ctx context.Context, tuples []language.Tuple) error +} + +func (m *mockTupleClient) ReadTuples(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) { + if m.readFn != nil { + return m.readFn(ctx, filter) + } + return nil, nil +} + +func (m *mockTupleClient) WriteTuples(ctx context.Context, tuples []language.Tuple) error { + if m.writeFn != nil { + return m.writeFn(ctx, tuples) + } + return nil +} + +func TestSyncMapping(t *testing.T) { + t.Parallel() + + t.Run("happy path processes two records", func(t *testing.T) { + t.Parallel() + + var called int + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + called++ + return nil + }, + } + + var out, errOut bytes.Buffer + err := syncMapping(context.Background(), "testdata/valid.yaml", syncOptions{}, + client, strings.NewReader("{\"id\":\"anne\"}\n{\"id\":\"bob\"}\n"), + &out, &errOut) + require.NoError(t, err) + assert.Equal(t, 2, called) + assert.Empty(t, out.String()) + assert.Contains(t, errOut.String(), "✓") + }) + + t.Run("quiet suppresses all progress output", func(t *testing.T) { + t.Parallel() + + var errOut bytes.Buffer + err := syncMapping(context.Background(), "testdata/valid.yaml", syncOptions{quiet: true}, + &mockTupleClient{}, strings.NewReader("{\"id\":\"anne\"}\n"), + &bytes.Buffer{}, &errOut) + require.NoError(t, err) + assert.Empty(t, errOut.String()) + }) + + t.Run("invalid mapping returns errMappingInvalid and writes diagnostics to errOut", func(t *testing.T) { + t.Parallel() + + var errOut bytes.Buffer + err := syncMapping(context.Background(), "testdata/invalid.yaml", syncOptions{}, + &mockTupleClient{}, strings.NewReader(""), + &bytes.Buffer{}, &errOut) + require.ErrorIs(t, err, errMappingInvalid) + assert.NotEmpty(t, errOut.String()) + }) + + t.Run("missing mapping file returns error containing 'reading'", func(t *testing.T) { + t.Parallel() + + err := syncMapping(context.Background(), "testdata/nonexistent.yaml", syncOptions{}, + &mockTupleClient{}, strings.NewReader(""), + &bytes.Buffer{}, &bytes.Buffer{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "reading") + }) + + t.Run("malformed JSON record returns error without continue-on-error", func(t *testing.T) { + t.Parallel() + + err := syncMapping(context.Background(), "testdata/valid.yaml", syncOptions{}, + &mockTupleClient{}, strings.NewReader("not json\n"), + &bytes.Buffer{}, &bytes.Buffer{}) + require.Error(t, err) + assert.False(t, errors.Is(err, errSyncRecordsFailed)) + assert.False(t, errors.Is(err, errMappingInvalid)) + }) + + t.Run("continue-on-error skips bad records and processes good ones", func(t *testing.T) { + t.Parallel() + + var called int + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + called++ + return nil + }, + } + + var errOut bytes.Buffer + err := syncMapping(context.Background(), "testdata/valid.yaml", + syncOptions{continueOnError: true}, + client, + strings.NewReader("not json\n{\"id\":\"anne\"}\n"), + &bytes.Buffer{}, &errOut) + require.ErrorIs(t, err, errSyncRecordsFailed) + assert.Equal(t, 1, called) + assert.Contains(t, errOut.String(), "✗") + }) + + t.Run("continue-on-error with all valid records returns nil", func(t *testing.T) { + t.Parallel() + + var called int + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + called++ + return nil + }, + } + + err := syncMapping(context.Background(), "testdata/valid.yaml", + syncOptions{continueOnError: true}, + client, + strings.NewReader("{\"id\":\"anne\"}\n{\"id\":\"bob\"}\n"), + &bytes.Buffer{}, &bytes.Buffer{}) + require.NoError(t, err) + assert.Equal(t, 2, called) + }) + + t.Run("dry-run does not call WriteTuples on the base client", func(t *testing.T) { + t.Parallel() + + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + t.Fatal("WriteTuples must not be called in dry-run") + return nil + }, + } + + err := syncMapping(context.Background(), "testdata/valid.yaml", + syncOptions{dryRun: true}, + client, + strings.NewReader("{\"id\":\"anne\"}\n"), + &bytes.Buffer{}, &bytes.Buffer{}) + require.NoError(t, err) + }) + + t.Run("dry-run emits JSONL to stdout when stdout is a non-TTY file", func(t *testing.T) { + t.Parallel() + + pr, pw, err := os.Pipe() + require.NoError(t, err) + defer pr.Close() + + syncErr := syncMapping(context.Background(), "testdata/valid.yaml", + syncOptions{dryRun: true}, + &mockTupleClient{}, + strings.NewReader("{\"id\":\"anne\"}\n"), + pw, &bytes.Buffer{}) + pw.Close() + require.NoError(t, syncErr) + + var buf bytes.Buffer + _, _ = io.Copy(&buf, pr) + assert.Contains(t, buf.String(), "user:anne") + }) + + t.Run("WriteTuples error is returned wrapped in WriteError", func(t *testing.T) { + t.Parallel() + + writeErr := errors.New("store unavailable") + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + return writeErr + }, + } + + err := syncMapping(context.Background(), "testdata/valid.yaml", syncOptions{}, + client, + strings.NewReader("{\"id\":\"anne\"}\n"), + &bytes.Buffer{}, &bytes.Buffer{}) + require.Error(t, err) + var wErr *apply.WriteError + assert.True(t, errors.As(err, &wErr)) + }) +} diff --git a/go.mod b/go.mod index 1e13230a..8e5712a9 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ toolchain go1.27.1 require ( github.com/charmbracelet/huh v1.0.0 + github.com/charmbracelet/lipgloss v1.1.0 github.com/mattn/go-isatty v0.0.24 github.com/muesli/mango-cobra v1.3.0 github.com/muesli/roff v0.1.0 @@ -14,7 +15,7 @@ require ( github.com/openfga/api/proto v0.0.0-20260319214821-f153694bfc20 github.com/openfga/go-sdk v0.8.3 github.com/openfga/language/pkg/go v0.3.2 - github.com/openfga/mapper v0.0.0-20260916114550-64ae7f46a687 + github.com/openfga/mapper v0.0.0-20261008093810-15f546b69fc9 github.com/openfga/openfga v1.21.0 github.com/schollz/progressbar/v3 v3.19.1 github.com/spf13/cobra v1.10.2 @@ -41,7 +42,6 @@ require ( github.com/charmbracelet/bubbles v0.21.1-0.20250623103423-23b8fd6302d7 // indirect github.com/charmbracelet/bubbletea v1.3.6 // indirect github.com/charmbracelet/colorprofile v0.2.3-0.20250311203215-f60798e515dc // indirect - github.com/charmbracelet/lipgloss v1.1.0 // indirect github.com/charmbracelet/x/ansi v0.9.3 // indirect github.com/charmbracelet/x/cellbuf v0.0.13 // indirect github.com/charmbracelet/x/exp/strings v0.0.0-20240722160745-212f7b056ed0 // indirect diff --git a/go.sum b/go.sum index d9e7af43..5454bbfc 100644 --- a/go.sum +++ b/go.sum @@ -235,8 +235,8 @@ github.com/openfga/go-sdk v0.8.3 h1:I4haME2GkR+gzmJY3vuGjuaLfdNyuMixsqj0lQbB4zE= github.com/openfga/go-sdk v0.8.3/go.mod h1:4EwBqOVHnIfCsQZ6e7LTbgr+jKfngu3YibPPV1mLXCc= github.com/openfga/language/pkg/go v0.3.2 h1:QN/cBYzL50RzEh46e+7ILvFeBs0exge7Sia0B6uXjms= github.com/openfga/language/pkg/go v0.3.2/go.mod h1:Ia3Luc0WcZdLogjjnFFKbce8+427wwlDaUG3dREZVCY= -github.com/openfga/mapper v0.0.0-20260916114550-64ae7f46a687 h1:Ioar5QWl0+PKBoP8034WNgsYqf8inNKLp6ZLw475zFs= -github.com/openfga/mapper v0.0.0-20260916114550-64ae7f46a687/go.mod h1:UdSK58M1OLJ1v+l3hElBFfeCuU3WncuG6H2pySWqNaY= +github.com/openfga/mapper v0.0.0-20261008093810-15f546b69fc9 h1:e131oeOLuT7rGfk8k2N2ib1Nb4omjDrnYH9dV9ADh6Q= +github.com/openfga/mapper v0.0.0-20261008093810-15f546b69fc9/go.mod h1:UdSK58M1OLJ1v+l3hElBFfeCuU3WncuG6H2pySWqNaY= github.com/openfga/openfga v1.21.0 h1:NQIrzLjUaWw+ZgFr78+qMcoYnHroPfqeQ8Z8ZMA530Y= github.com/openfga/openfga v1.21.0/go.mod h1:yO9enSlCudWGsiuYgD3DZDHwZJTlz3hSiyUCda8NGfk= github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= diff --git a/tests/fixtures/sync-event.json b/tests/fixtures/sync-event.json new file mode 100644 index 00000000..3d008820 --- /dev/null +++ b/tests/fixtures/sync-event.json @@ -0,0 +1 @@ +{"id":"anne","group":"eng"} diff --git a/tests/fixtures/sync-mapping.yaml b/tests/fixtures/sync-mapping.yaml new file mode 100644 index 00000000..ecfb5803 --- /dev/null +++ b/tests/fixtures/sync-mapping.yaml @@ -0,0 +1,7 @@ +version: "1" +rules: + - name: "add-owner" + tuples: + - user: "user:{{ input.id }}" + relation: "owner" + object: "group:{{ input.group }}" diff --git a/tests/mapping-sync-test-cases.yaml b/tests/mapping-sync-test-cases.yaml new file mode 100644 index 00000000..9900dcf8 --- /dev/null +++ b/tests/mapping-sync-test-cases.yaml @@ -0,0 +1,54 @@ +config: + inherit-env: true + +tests: + # fga mapping sync — store-independent error cases + + 001 - sync missing store-id flag exits non-zero: + command: bash -c 'echo "{}" | fga mapping sync ./tests/fixtures/sync-mapping.yaml' + exit-code: 1 + + 002 - sync invalid mapping exits 2 with diagnostics on stderr: + command: bash -c 'echo "{}" | fga mapping sync ./tests/fixtures/invalid-mapping.yaml --store-id=$(./tests/scripts/test-data/get-store-id.sh)' + exit-code: 2 + stderr: + contains: + - "unclosed" + + # fga mapping sync — store-backed cases + + 003 - sync writes a tuple and exits 0: + command: bash -c 'cat ./tests/fixtures/sync-event.json | fga mapping sync ./tests/fixtures/sync-mapping.yaml --store-id=$(./tests/scripts/test-data/get-store-id.sh) --model-id=$(./tests/scripts/test-data/get-model-id.sh)' + exit-code: 0 + stderr: + contains: + - "✓" + + 004 - sync --dry-run prints JSONL to stdout without writing: + command: bash -c 'cat ./tests/fixtures/sync-event.json | fga mapping sync ./tests/fixtures/sync-mapping.yaml --store-id=$(./tests/scripts/test-data/get-store-id.sh) --model-id=$(./tests/scripts/test-data/get-model-id.sh) --dry-run' + exit-code: 0 + stdout: + json: + user: "user:anne" + relation: "owner" + object: "group:eng" + op: "write" + + 005 - sync --quiet suppresses progress output: + command: bash -c 'cat ./tests/fixtures/sync-event.json | fga mapping sync ./tests/fixtures/sync-mapping.yaml --store-id=$(./tests/scripts/test-data/get-store-id.sh) --model-id=$(./tests/scripts/test-data/get-model-id.sh) --quiet' + exit-code: 0 + stderr: + not-contains: + - "✓" + - "records" + + 006 - sync --continue-on-error skips bad records and exits 1: + command: bash -c 'printf "bad json\n{\"id\":\"anne\",\"group\":\"eng\"}\n" | fga mapping sync ./tests/fixtures/sync-mapping.yaml --store-id=$(./tests/scripts/test-data/get-store-id.sh) --model-id=$(./tests/scripts/test-data/get-model-id.sh) --continue-on-error' + exit-code: 1 + stderr: + contains: + - "✗" + + 007 - sync --input reads from file and exits 0: + command: fga mapping sync ./tests/fixtures/sync-mapping.yaml --input ./tests/fixtures/sync-event.json --store-id=$(./tests/scripts/test-data/get-store-id.sh) --model-id=$(./tests/scripts/test-data/get-model-id.sh) + exit-code: 0 From 328c5c0151434761aeda82a5a198a82f8179379f Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 10:53:34 +0100 Subject: [PATCH 2/6] docs: document fga mapping sync in README --- README.md | 43 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 41 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 52ca2b69..888c3388 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,7 @@ A cross-platform CLI to interact with an OpenFGA server - [Run Embedded Tests](#test-mapping) - [Scaffold a Mapping File](#init-mapping) - [Evaluate Against JSON Input](#run-mapping) + - [Sync to Store](#sync-mapping) - [Contributing](#contributing) - [License](#license) @@ -1217,8 +1218,7 @@ fga query **list-users** --object --relation --user-filter < #### Mapping -Offline tooling for authoring, validating, testing, and evaluating JSON→tuple mapping files. -No store credentials or network access required. +Tooling for authoring, validating, testing, evaluating, and applying JSON→tuple mapping files. Most commands run offline; `sync` requires store credentials. - `mapping` @@ -1227,6 +1227,7 @@ No store credentials or network access required. | [Scaffold a mapping file](#init-mapping) | `init` | `[mapping.yaml]`, `--minimal`, `--force` | `fga mapping init` | | [Validate a mapping file](#validate-mapping) | `validate` | `--format`, `--model-file`, `--verbose` | `fga mapping validate mapping.yaml` | | [Run embedded tests](#test-mapping) | `test` | `--format`, `--run`, `--fail-fast`, `--output-file`, `--verbose`, `--no-color` | `fga mapping test mapping.yaml` | +| [Sync to store](#sync-mapping) | `sync` | `--store-id`, `--model-id`, `--input`, `--dry-run`, `--continue-on-error`, `--quiet` | `fga mapping sync mapping.yaml --store-id $FGA_STORE_ID < events.jsonl` | ##### Validate Mapping @@ -1357,6 +1358,44 @@ Error: invalid JSON at line 1, column 8: invalid character 'b' looking for begin > :quit ``` +##### Sync Mapping + +###### Command +fga mapping **sync** [mapping-file] + +Reads JSONL from stdin (or `--input`) and evaluates each record against the mapping file, then applies the resulting tuple writes and deletes to the store. Rules using `tuple_filters` read current store state, diff against desired state, and derive the minimal write/delete set before applying. Deletes are issued before writes so that a condition change on the same user–relation–object (delete old + write new) lands correctly. + +The mapping file is optional: omit it in an interactive terminal to choose one from a `.yaml`/`.yml` file picker. + +###### Parameters +* `--store-id` (required): Store ID +* `--model-id`: Authorization Model ID +* `--input`: Path to a JSONL input file, one JSON object per line (default: stdin) +* `--dry-run`: Read the store and print planned operations to stdout as JSONL instead of writing them. Suppressed at an interactive terminal where the output would be unreadable. +* `--continue-on-error`: Skip records that fail to evaluate or apply (shown on stderr) and exit non-zero if any were skipped +* `--quiet`: Suppress progress output; only errors are written to stderr + +###### Example +`echo '{"id":"anne","org":"acme"}' | fga mapping sync mapping.yaml --store-id $FGA_STORE_ID` + +`fga mapping sync mapping.yaml --input events.jsonl --store-id $FGA_STORE_ID --dry-run > planned.jsonl` + +`fga mapping sync mapping.yaml --store-id $FGA_STORE_ID --continue-on-error < events.jsonl` + +###### Response + +Progress is shown on stderr as a single updating line (or once per second when not a terminal): +``` +15:04:05 ✓ +12 -3 10 records +``` +With `--quiet`, only errors appear on stderr. + +`--dry-run` (with stdout redirected) emits planned operations as JSONL: +```json +{"op":"write","user":"user:anne","relation":"member","object":"org:acme"} +{"op":"delete","user":"user:bob","relation":"member","object":"org:acme"} +``` + ## Contributing See [CONTRIBUTING](https://github.com/openfga/.github/blob/main/CONTRIBUTING.md). From 157cde3e41018f72b1a3f5b37f238c7f55f17556 Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 10:58:14 +0100 Subject: [PATCH 3/6] fix(lint): add lipgloss to depguard allowlist, drop unnecessary io.Writer cast --- .golangci.yaml | 1 + cmd/mapping/sync.go | 25 ++++++++++++------------- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/.golangci.yaml b/.golangci.yaml index 4c331193..ea95c75d 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -42,6 +42,7 @@ linters: allow: - $gostd - github.com/charmbracelet/huh + - github.com/charmbracelet/lipgloss - github.com/gocarina/gocsv - github.com/hashicorp/go-multierror - github.com/mattn/go-isatty diff --git a/cmd/mapping/sync.go b/cmd/mapping/sync.go index 926838cd..34375a90 100644 --- a/cmd/mapping/sync.go +++ b/cmd/mapping/sync.go @@ -24,22 +24,22 @@ var errSyncRecordsFailed = errors.New("one or more records failed") // errStyles holds lipgloss styles bound to a specific output writer. type errStyles struct { - time lipgloss.Style - writes lipgloss.Style - deletes lipgloss.Style - dim lipgloss.Style - ok lipgloss.Style + time lipgloss.Style + writes lipgloss.Style + deletes lipgloss.Style + dim lipgloss.Style + ok lipgloss.Style errStyle lipgloss.Style } func newErrStyles(w io.Writer) errStyles { r := lipgloss.NewRenderer(w) return errStyles{ - time: r.NewStyle().Faint(true), - writes: r.NewStyle().Foreground(lipgloss.Color("10")).Bold(true), - deletes: r.NewStyle().Foreground(lipgloss.Color("9")).Bold(true), - dim: r.NewStyle().Faint(true), - ok: r.NewStyle().Foreground(lipgloss.Color("10")), + time: r.NewStyle().Faint(true), + writes: r.NewStyle().Foreground(lipgloss.Color("10")).Bold(true), + deletes: r.NewStyle().Foreground(lipgloss.Color("9")).Bold(true), + dim: r.NewStyle().Faint(true), + ok: r.NewStyle().Foreground(lipgloss.Color("10")), errStyle: r.NewStyle().Foreground(lipgloss.Color("9")), } } @@ -154,9 +154,9 @@ func syncMapping( // For dry-run, wrap the base client so writes are printed instead of applied. // Only emit JSONL when stdout is redirected — at a TTY it's unreadable at scale. - var writeTarget apply.TupleClient = base + writeTarget := base if opts.dryRun { - dryOut := io.Writer(io.Discard) + dryOut := io.Discard if f, ok := out.(*os.File); ok && !isatty.IsTerminal(f.Fd()) { dryOut = out } @@ -289,7 +289,6 @@ func processSyncRecord( return cc.writes, cc.dels, nil } - var ( syncInputFile string syncDryRun bool From dda900e591cba3344da2cfa4190866f541057501 Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 13:13:55 +0100 Subject: [PATCH 4/6] fix(mapping): address PR review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract printer types from sync.go into printer.go to keep sync logic separate from presentation. Narrow fgaTupleClient.inner and internaltuple.Read's parameter to minimal interfaces for testability. Pass HIGHER_CONSISTENCY on reads in ReadTuples to avoid stale-replica data during a sync diff. Fix a goroutine race where the worker could process the next record after sending an error when continueOnError is false — it now returns immediately after sending. Unset FGA_STORE_ID in the missing-store-id integration test to prevent the ambient env var from masking the failure. Add tests covering HIGHER_CONSISTENCY propagation, per-tuple FAILURE on deletes, and the goroutine early-return behaviour. --- cmd/mapping/client.go | 11 +- cmd/mapping/client_test.go | 173 +++++++++++++++++++++++++++++ cmd/mapping/printer.go | 115 +++++++++++++++++++ cmd/mapping/sync.go | 109 +----------------- cmd/mapping/sync_test.go | 19 ++++ internal/tuple/read.go | 7 +- tests/mapping-sync-test-cases.yaml | 2 +- 7 files changed, 326 insertions(+), 110 deletions(-) create mode 100644 cmd/mapping/printer.go diff --git a/cmd/mapping/client.go b/cmd/mapping/client.go index 321e5153..36376eb2 100644 --- a/cmd/mapping/client.go +++ b/cmd/mapping/client.go @@ -17,9 +17,15 @@ import ( const maxWriteChunkSize = 40 +// sdkWriteReadClient is the subset of sdkclient.SdkClient used by fgaTupleClient. +type sdkWriteReadClient interface { + Read(ctx context.Context) sdkclient.SdkClientReadRequestInterface + Write(ctx context.Context) sdkclient.SdkClientWriteRequestInterface +} + // fgaTupleClient adapts the OpenFGA SDK client to the apply.TupleClient interface. type fgaTupleClient struct { - inner sdkclient.SdkClient + inner sdkWriteReadClient } func (c *fgaTupleClient) ReadTuples(ctx context.Context, filter language.TupleFilter) ([]language.Tuple, error) { @@ -34,7 +40,8 @@ func (c *fgaTupleClient) ReadTuples(ctx context.Context, filter language.TupleFi body.Object = &filter.Object } - resp, err := internaltuple.Read(ctx, c.inner, body, 0, internaltuple.DefaultReadPageSize, nil) + consistency := openfga.CONSISTENCYPREFERENCE_HIGHER_CONSISTENCY + resp, err := internaltuple.Read(ctx, c.inner, body, 0, internaltuple.DefaultReadPageSize, &consistency) if err != nil { return nil, err } diff --git a/cmd/mapping/client_test.go b/cmd/mapping/client_test.go index 491c2a09..e1bbbed5 100644 --- a/cmd/mapping/client_test.go +++ b/cmd/mapping/client_test.go @@ -7,11 +7,94 @@ import ( "strings" "testing" + openfga "github.com/openfga/go-sdk" + sdkclient "github.com/openfga/go-sdk/client" "github.com/openfga/mapper/language" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// mockReadRequest implements sdkclient.SdkClientReadRequestInterface. +// It captures the Options call so tests can assert consistency preference. +type mockReadRequest struct { + capturedOptions sdkclient.ClientReadOptions + tuples []openfga.Tuple + err error +} + +func (m *mockReadRequest) Options(opts sdkclient.ClientReadOptions) sdkclient.SdkClientReadRequestInterface { + m.capturedOptions = opts + return m +} + +func (m *mockReadRequest) Body(_ sdkclient.ClientReadRequest) sdkclient.SdkClientReadRequestInterface { + return m +} + +func (m *mockReadRequest) Execute() (*openfga.ReadResponse, error) { + if m.err != nil { + return nil, m.err + } + return &openfga.ReadResponse{Tuples: m.tuples}, nil +} + +func (m *mockReadRequest) GetStoreIdOverride() *string { return nil } +func (m *mockReadRequest) GetContext() context.Context { return context.Background() } +func (m *mockReadRequest) GetBody() *sdkclient.ClientReadRequest { return nil } +func (m *mockReadRequest) GetOptions() *sdkclient.ClientReadOptions { return nil } + +// mockWriteRequest implements sdkclient.SdkClientWriteRequestInterface. +// It captures the Body and returns a configured response from Execute. +type mockWriteRequest struct { + capturedBody sdkclient.ClientWriteRequest + response *sdkclient.ClientWriteResponse + err error +} + +func (m *mockWriteRequest) Options(_ sdkclient.ClientWriteOptions) sdkclient.SdkClientWriteRequestInterface { + return m +} + +func (m *mockWriteRequest) Body(b sdkclient.ClientWriteRequest) sdkclient.SdkClientWriteRequestInterface { + m.capturedBody = b + return m +} + +func (m *mockWriteRequest) Execute() (*sdkclient.ClientWriteResponse, error) { + if m.err != nil { + return nil, m.err + } + if m.response != nil { + return m.response, nil + } + return &sdkclient.ClientWriteResponse{}, nil +} + +func (m *mockWriteRequest) GetAuthorizationModelIdOverride() *string { return nil } +func (m *mockWriteRequest) GetStoreIdOverride() *string { return nil } +func (m *mockWriteRequest) GetContext() context.Context { return context.Background() } +func (m *mockWriteRequest) GetOptions() *sdkclient.ClientWriteOptions { return nil } +func (m *mockWriteRequest) GetBody() *sdkclient.ClientWriteRequest { return nil } + +// mockSdkClient implements sdkWriteReadClient. +// writeReqFn, when set, is called on each Write() invocation; otherwise writeReq is returned. +type mockSdkClient struct { + readReq *mockReadRequest + writeReq *mockWriteRequest + writeReqFn func() sdkclient.SdkClientWriteRequestInterface +} + +func (m *mockSdkClient) Read(_ context.Context) sdkclient.SdkClientReadRequestInterface { + return m.readReq +} + +func (m *mockSdkClient) Write(_ context.Context) sdkclient.SdkClientWriteRequestInterface { + if m.writeReqFn != nil { + return m.writeReqFn() + } + return m.writeReq +} + type alwaysErrWriter struct{} func (e *alwaysErrWriter) Write(_ []byte) (int, error) { @@ -203,3 +286,93 @@ func TestPartitionSyncTuples(t *testing.T) { assert.Nil(t, deletes) }) } + +func TestFGATupleClientReadTuples(t *testing.T) { + t.Parallel() + + t.Run("passes HIGHER_CONSISTENCY and filter to Read", func(t *testing.T) { + t.Parallel() + + existing := openfga.Tuple{Key: openfga.TupleKey{ + User: "user:anne", + Relation: "admin", + Object: "org:acme", + }} + readReq := &mockReadRequest{tuples: []openfga.Tuple{existing}} + fc := &fgaTupleClient{inner: &mockSdkClient{readReq: readReq}} + + filter := language.TupleFilter{User: "user:anne", Object: "org:acme"} + tuples, err := fc.ReadTuples(context.Background(), filter) + + require.NoError(t, err) + require.Len(t, tuples, 1) + assert.Equal(t, "user:anne", tuples[0].User) + assert.Equal(t, "admin", tuples[0].Relation) + require.NotNil(t, readReq.capturedOptions.Consistency) + assert.Equal(t, openfga.CONSISTENCYPREFERENCE_HIGHER_CONSISTENCY, *readReq.capturedOptions.Consistency) + }) +} + +func TestFGATupleClientWriteTuples(t *testing.T) { + t.Parallel() + + t.Run("per-tuple FAILURE on delete propagates as error", func(t *testing.T) { + t.Parallel() + + writeReq := &mockWriteRequest{ + response: &sdkclient.ClientWriteResponse{ + Deletes: []sdkclient.ClientWriteRequestDeleteResponse{ + { + TupleKey: sdkclient.ClientTupleKeyWithoutCondition{ + User: "user:anne", + Relation: "member", + Object: "org:acme", + }, + Status: sdkclient.FAILURE, + }, + }, + }, + } + fc := &fgaTupleClient{inner: &mockSdkClient{writeReq: writeReq}} + + err := fc.WriteTuples(context.Background(), []language.Tuple{ + {User: "user:anne", Relation: "member", Object: "org:acme", Action: language.ActionDelete}, + }) + + require.Error(t, err) + assert.Contains(t, err.Error(), "deleting tuples") + }) + + t.Run("per-tuple FAILURE on delete prevents writes", func(t *testing.T) { + t.Parallel() + + var writeCalls int + sdk := &mockSdkClient{ + writeReqFn: func() sdkclient.SdkClientWriteRequestInterface { + writeCalls++ + if writeCalls > 1 { + t.Error("Write must not be called after a delete failure") + } + return &mockWriteRequest{ + response: &sdkclient.ClientWriteResponse{ + Deletes: []sdkclient.ClientWriteRequestDeleteResponse{ + { + Status: sdkclient.FAILURE, + TupleKey: sdkclient.ClientTupleKeyWithoutCondition{User: "user:anne", Relation: "admin", Object: "org:acme"}, + }, + }, + }, + } + }, + } + fc := &fgaTupleClient{inner: sdk} + + err := fc.WriteTuples(context.Background(), []language.Tuple{ + {User: "user:anne", Relation: "admin", Object: "org:acme", Action: language.ActionDelete}, + {User: "user:anne", Relation: "viewer", Object: "org:acme", Action: language.ActionWrite}, + }) + + require.Error(t, err) + assert.Equal(t, 1, writeCalls, "Write must not be called a second time after delete failure") + }) +} diff --git a/cmd/mapping/printer.go b/cmd/mapping/printer.go new file mode 100644 index 00000000..784491f9 --- /dev/null +++ b/cmd/mapping/printer.go @@ -0,0 +1,115 @@ +package mapping + +import ( + "fmt" + "io" + "os" + "time" + + "github.com/charmbracelet/lipgloss" + "github.com/mattn/go-isatty" +) + +type errStyles struct { + time lipgloss.Style + writes lipgloss.Style + deletes lipgloss.Style + dim lipgloss.Style + ok lipgloss.Style + errStyle lipgloss.Style +} + +func newErrStyles(w io.Writer) errStyles { + r := lipgloss.NewRenderer(w) + return errStyles{ + time: r.NewStyle().Faint(true), + writes: r.NewStyle().Foreground(lipgloss.Color("10")).Bold(true), + deletes: r.NewStyle().Foreground(lipgloss.Color("9")).Bold(true), + dim: r.NewStyle().Faint(true), + ok: r.NewStyle().Foreground(lipgloss.Color("10")), + errStyle: r.NewStyle().Foreground(lipgloss.Color("9")), + } +} + +var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} + +type syncStats struct { + records int + writes int + deletes int + errors int + frame int +} + +type statusPrinter struct { + out io.Writer + isTTY bool + quiet bool + styles errStyles + lastRender time.Time +} + +func newStatusPrinter(out io.Writer, quiet bool) *statusPrinter { + isTTY := false + if f, ok := out.(*os.File); ok { + isTTY = isatty.IsTerminal(f.Fd()) + } + return &statusPrinter{out: out, isTTY: isTTY, quiet: quiet, styles: newErrStyles(out)} +} + +func (p *statusPrinter) render(stats syncStats, suffix string) { + if p.quiet { + return + } + + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + spin := p.styles.dim.Render(spinnerFrames[stats.frame%len(spinnerFrames)]) + w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) + d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) + count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) + + line := fmt.Sprintf("%s %s %s %s %s", ts, spin, w, d, count) + if suffix != "" { + line += " " + suffix + } + + if p.isTTY { + fmt.Fprintf(p.out, "\r%s\x1b[K", line) + } else if time.Since(p.lastRender) >= time.Second { + p.lastRender = time.Now() + fmt.Fprintln(p.out, line) + } +} + +func (p *statusPrinter) done(stats syncStats) { + if p.quiet { + return + } + + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + tick := p.styles.ok.Render("✓") + w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) + d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) + count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) + + line := fmt.Sprintf("%s %s %s %s %s", ts, tick, w, d, count) + if stats.errors > 0 { + line += " " + p.styles.errStyle.Render(fmt.Sprintf("%d errors", stats.errors)) + } + + if p.isTTY { + fmt.Fprintf(p.out, "\r%s\x1b[K\n", line) + } else { + fmt.Fprintln(p.out, line) + } +} + +func (p *statusPrinter) errLine(msg string) { + ts := p.styles.time.Render(time.Now().Format("15:04:05")) + x := p.styles.errStyle.Render("✗") + if p.isTTY { + fmt.Fprintf(p.out, "\r%s %s %s\x1b[K\n", ts, x, msg) + } else { + fmt.Fprintf(p.out, "%s %s %s\n", ts, x, msg) + } +} diff --git a/cmd/mapping/sync.go b/cmd/mapping/sync.go index 34375a90..7e16323f 100644 --- a/cmd/mapping/sync.go +++ b/cmd/mapping/sync.go @@ -11,7 +11,6 @@ import ( "strings" "time" - "github.com/charmbracelet/lipgloss" "github.com/mattn/go-isatty" "github.com/openfga/mapper" "github.com/openfga/mapper/apply" @@ -22,117 +21,12 @@ import ( var errSyncRecordsFailed = errors.New("one or more records failed") -// errStyles holds lipgloss styles bound to a specific output writer. -type errStyles struct { - time lipgloss.Style - writes lipgloss.Style - deletes lipgloss.Style - dim lipgloss.Style - ok lipgloss.Style - errStyle lipgloss.Style -} - -func newErrStyles(w io.Writer) errStyles { - r := lipgloss.NewRenderer(w) - return errStyles{ - time: r.NewStyle().Faint(true), - writes: r.NewStyle().Foreground(lipgloss.Color("10")).Bold(true), - deletes: r.NewStyle().Foreground(lipgloss.Color("9")).Bold(true), - dim: r.NewStyle().Faint(true), - ok: r.NewStyle().Foreground(lipgloss.Color("10")), - errStyle: r.NewStyle().Foreground(lipgloss.Color("9")), - } -} - -var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} - type syncOptions struct { quiet bool dryRun bool continueOnError bool } -type syncStats struct { - records int - writes int - deletes int - errors int - frame int -} - -type statusPrinter struct { - out io.Writer - isTTY bool - quiet bool - styles errStyles - lastRender time.Time -} - -func newStatusPrinter(out io.Writer, quiet bool) *statusPrinter { - isTTY := false - if f, ok := out.(*os.File); ok { - isTTY = isatty.IsTerminal(f.Fd()) - } - return &statusPrinter{out: out, isTTY: isTTY, quiet: quiet, styles: newErrStyles(out)} -} - -func (p *statusPrinter) render(stats syncStats, suffix string) { - if p.quiet { - return - } - - ts := p.styles.time.Render(time.Now().Format("15:04:05")) - spin := p.styles.dim.Render(spinnerFrames[stats.frame%len(spinnerFrames)]) - w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) - d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) - count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) - - line := fmt.Sprintf("%s %s %s %s %s", ts, spin, w, d, count) - if suffix != "" { - line += " " + suffix - } - - if p.isTTY { - fmt.Fprintf(p.out, "\r%s\x1b[K", line) - } else if time.Since(p.lastRender) >= time.Second { - p.lastRender = time.Now() - fmt.Fprintln(p.out, line) - } -} - -func (p *statusPrinter) done(stats syncStats) { - if p.quiet { - return - } - - ts := p.styles.time.Render(time.Now().Format("15:04:05")) - tick := p.styles.ok.Render("✓") - w := p.styles.writes.Render(fmt.Sprintf("+%d", stats.writes)) - d := p.styles.deletes.Render(fmt.Sprintf("-%d", stats.deletes)) - count := p.styles.dim.Render(fmt.Sprintf("%d records", stats.records)) - - line := fmt.Sprintf("%s %s %s %s %s", ts, tick, w, d, count) - if stats.errors > 0 { - line += " " + p.styles.errStyle.Render(fmt.Sprintf("%d errors", stats.errors)) - } - - if p.isTTY { - fmt.Fprintf(p.out, "\r%s\x1b[K\n", line) - } else { - fmt.Fprintln(p.out, line) - } -} - -func (p *statusPrinter) errLine(msg string) { - ts := p.styles.time.Render(time.Now().Format("15:04:05")) - x := p.styles.errStyle.Render("✗") - if p.isTTY { - fmt.Fprintf(p.out, "\r%s %s %s\x1b[K\n", ts, x, msg) - } else { - fmt.Fprintf(p.out, "%s %s %s\n", ts, x, msg) - } -} - func syncMapping( ctx context.Context, path string, @@ -197,6 +91,9 @@ func syncMapping( w, d, recErr := processSyncRecord(runCtx, compiled, rec, cc, trimmed) select { case results <- recordResult{writes: w, deletes: d, err: recErr}: + if recErr != nil && !opts.continueOnError { + return + } case <-runCtx.Done(): return } diff --git a/cmd/mapping/sync_test.go b/cmd/mapping/sync_test.go index db9af698..e84c8b2c 100644 --- a/cmd/mapping/sync_test.go +++ b/cmd/mapping/sync_test.go @@ -181,6 +181,25 @@ func TestSyncMapping(t *testing.T) { assert.Contains(t, buf.String(), "user:anne") }) + t.Run("error without continue-on-error does not write subsequent records", func(t *testing.T) { + t.Parallel() + + var called int + client := &mockTupleClient{ + writeFn: func(_ context.Context, _ []language.Tuple) error { + called++ + return nil + }, + } + + err := syncMapping(context.Background(), "testdata/valid.yaml", syncOptions{}, + client, + strings.NewReader("not json\n{\"id\":\"anne\"}\n"), + &bytes.Buffer{}, &bytes.Buffer{}) + require.Error(t, err) + assert.Equal(t, 0, called, "valid record after failure must not be written") + }) + t.Run("WriteTuples error is returned wrapped in WriteError", func(t *testing.T) { t.Parallel() diff --git a/internal/tuple/read.go b/internal/tuple/read.go index 48dca088..166e408b 100644 --- a/internal/tuple/read.go +++ b/internal/tuple/read.go @@ -14,9 +14,14 @@ const DefaultReadPageSize int32 = 50 // ErrInvalidPageSize is returned when page size is outside valid range. var ErrInvalidPageSize = errors.New("pageSize must be between 1 and 100") +// TupleReaderClient is the subset of client.SdkClient required by Read. +type TupleReaderClient interface { + Read(ctx context.Context) client.SdkClientReadRequestInterface +} + func Read( ctx context.Context, - fgaClient client.SdkClient, + fgaClient TupleReaderClient, body *client.ClientReadRequest, maxPages int, pageSize int32, diff --git a/tests/mapping-sync-test-cases.yaml b/tests/mapping-sync-test-cases.yaml index 9900dcf8..55012059 100644 --- a/tests/mapping-sync-test-cases.yaml +++ b/tests/mapping-sync-test-cases.yaml @@ -5,7 +5,7 @@ tests: # fga mapping sync — store-independent error cases 001 - sync missing store-id flag exits non-zero: - command: bash -c 'echo "{}" | fga mapping sync ./tests/fixtures/sync-mapping.yaml' + command: env -u FGA_STORE_ID bash -c 'echo "{}" | fga mapping sync ./tests/fixtures/sync-mapping.yaml' exit-code: 1 002 - sync invalid mapping exits 2 with diagnostics on stderr: From fb461487bd328a845753504ab975e6b38059e71f Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 13:26:40 +0100 Subject: [PATCH 5/6] fix(lint): suppress staticcheck ST1003 on SDK interface method stubs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OpenFGA Go SDK interfaces use GetStoreIdOverride and GetAuthorizationModelIdOverride — names that predate Go naming conventions. Test mocks must implement the interface as-is, so suppress ST1003 on those stubs rather than using names that break the interface. --- cmd/mapping/client_test.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/cmd/mapping/client_test.go b/cmd/mapping/client_test.go index e1bbbed5..afb12a42 100644 --- a/cmd/mapping/client_test.go +++ b/cmd/mapping/client_test.go @@ -38,7 +38,7 @@ func (m *mockReadRequest) Execute() (*openfga.ReadResponse, error) { return &openfga.ReadResponse{Tuples: m.tuples}, nil } -func (m *mockReadRequest) GetStoreIdOverride() *string { return nil } +func (m *mockReadRequest) GetStoreIdOverride() *string { return nil } //nolint:staticcheck func (m *mockReadRequest) GetContext() context.Context { return context.Background() } func (m *mockReadRequest) GetBody() *sdkclient.ClientReadRequest { return nil } func (m *mockReadRequest) GetOptions() *sdkclient.ClientReadOptions { return nil } @@ -70,11 +70,11 @@ func (m *mockWriteRequest) Execute() (*sdkclient.ClientWriteResponse, error) { return &sdkclient.ClientWriteResponse{}, nil } -func (m *mockWriteRequest) GetAuthorizationModelIdOverride() *string { return nil } -func (m *mockWriteRequest) GetStoreIdOverride() *string { return nil } -func (m *mockWriteRequest) GetContext() context.Context { return context.Background() } -func (m *mockWriteRequest) GetOptions() *sdkclient.ClientWriteOptions { return nil } -func (m *mockWriteRequest) GetBody() *sdkclient.ClientWriteRequest { return nil } +func (m *mockWriteRequest) GetAuthorizationModelIdOverride() *string { return nil } //nolint:staticcheck +func (m *mockWriteRequest) GetStoreIdOverride() *string { return nil } //nolint:staticcheck +func (m *mockWriteRequest) GetContext() context.Context { return context.Background() } +func (m *mockWriteRequest) GetOptions() *sdkclient.ClientWriteOptions { return nil } +func (m *mockWriteRequest) GetBody() *sdkclient.ClientWriteRequest { return nil } // mockSdkClient implements sdkWriteReadClient. // writeReqFn, when set, is called on each Write() invocation; otherwise writeReq is returned. From ffae58a5d89ecab04e5e90be7a1a4781cee55707 Mon Sep 17 00:00:00 2001 From: Ewan Harris Date: Thu, 8 Oct 2026 16:47:06 +0100 Subject: [PATCH 6/6] refactor(mapping): extract executeDeletes/executeWrites and simplify partitionSyncTuples Extract the delete and write batch execution blocks from WriteTuples into dedicated executeDeletes and executeWrites methods to reduce cyclomatic complexity and make each path easier to follow in isolation. Simplify the partitionSyncTuples loop to use continue on the delete branch rather than an else block. --- cmd/mapping/client.go | 144 +++++++++++++++++++++++++----------------- 1 file changed, 86 insertions(+), 58 deletions(-) diff --git a/cmd/mapping/client.go b/cmd/mapping/client.go index 36376eb2..2dfce11e 100644 --- a/cmd/mapping/client.go +++ b/cmd/mapping/client.go @@ -87,60 +87,86 @@ func (c *fgaTupleClient) WriteTuples(ctx context.Context, tuples []language.Tupl // land first. The SDK with Transaction.Disable sends writes first then deletes, so we // split them into two calls. if len(deletes) > 0 { - resp, err := c.inner.Write(ctx).Options(writeOpts).Body(sdkclient.ClientWriteRequest{ - Deletes: deletes, - }).Execute() - if err != nil { - return fmt.Errorf("deleting tuples: %w", err) - } - var errs []error - seen := make(map[string]struct{}) - for _, d := range resp.Deletes { - if d.Status == sdkclient.FAILURE { - var e error - if d.Error != nil { - e = d.Error - } else { - e = fmt.Errorf("delete failed: %s %s %s", d.TupleKey.User, d.TupleKey.Relation, d.TupleKey.Object) - } - if _, ok := seen[e.Error()]; !ok { - seen[e.Error()] = struct{}{} - errs = append(errs, e) - } - } - } - if len(errs) > 0 { - return fmt.Errorf("deleting tuples: %w", errors.Join(errs...)) + if err := c.executeDeletes(ctx, writeOpts, deletes); err != nil { + return err } } if len(writes) > 0 { - resp, err := c.inner.Write(ctx).Options(writeOpts).Body(sdkclient.ClientWriteRequest{ - Writes: writes, - }).Execute() - if err != nil { - return fmt.Errorf("writing tuples: %w", err) + if err := c.executeWrites(ctx, writeOpts, writes); err != nil { + return err } - var errs []error - seen := make(map[string]struct{}) - for _, w := range resp.Writes { - if w.Status == sdkclient.FAILURE { - var e error - if w.Error != nil { - e = w.Error - } else { - e = fmt.Errorf("write failed: %s %s %s", w.TupleKey.User, w.TupleKey.Relation, w.TupleKey.Object) - } - if _, ok := seen[e.Error()]; !ok { - seen[e.Error()] = struct{}{} - errs = append(errs, e) - } + } + + return nil +} + +func (c *fgaTupleClient) executeDeletes( + ctx context.Context, + opts sdkclient.ClientWriteOptions, + deletes []sdkclient.ClientTupleKeyWithoutCondition, +) error { + resp, err := c.inner.Write(ctx).Options(opts).Body(sdkclient.ClientWriteRequest{ + Deletes: deletes, + }).Execute() + if err != nil { + return fmt.Errorf("deleting tuples: %w", err) + } + + var errs []error + seen := make(map[string]struct{}) + for _, d := range resp.Deletes { + if d.Status == sdkclient.FAILURE { + var e error + if d.Error != nil { + e = d.Error + } else { + e = fmt.Errorf("delete failed: %s %s %s", d.TupleKey.User, d.TupleKey.Relation, d.TupleKey.Object) + } + if _, ok := seen[e.Error()]; !ok { + seen[e.Error()] = struct{}{} + errs = append(errs, e) } } - if len(errs) > 0 { - return fmt.Errorf("writing tuples: %w", errors.Join(errs...)) + } + if len(errs) > 0 { + return fmt.Errorf("deleting tuples: %w", errors.Join(errs...)) + } + + return nil +} + +func (c *fgaTupleClient) executeWrites( + ctx context.Context, + opts sdkclient.ClientWriteOptions, + writes []sdkclient.ClientTupleKey, +) error { + resp, err := c.inner.Write(ctx).Options(opts).Body(sdkclient.ClientWriteRequest{ + Writes: writes, + }).Execute() + if err != nil { + return fmt.Errorf("writing tuples: %w", err) + } + + var errs []error + seen := make(map[string]struct{}) + for _, w := range resp.Writes { + if w.Status == sdkclient.FAILURE { + var e error + if w.Error != nil { + e = w.Error + } else { + e = fmt.Errorf("write failed: %s %s %s", w.TupleKey.User, w.TupleKey.Relation, w.TupleKey.Object) + } + if _, ok := seen[e.Error()]; !ok { + seen[e.Error()] = struct{}{} + errs = append(errs, e) + } } } + if len(errs) > 0 { + return fmt.Errorf("writing tuples: %w", errors.Join(errs...)) + } return nil } @@ -201,21 +227,23 @@ func partitionSyncTuples(tuples []language.Tuple) ([]sdkclient.ClientTupleKey, [ Relation: t.Relation, Object: t.Object, }) - } else { - key := sdkclient.ClientTupleKey{ - User: t.User, - Relation: t.Relation, - Object: t.Object, - } - if t.Condition != "" { - rc := &openfga.RelationshipCondition{Name: t.Condition} - if len(t.Context) > 0 { - rc.Context = &t.Context - } - key.Condition = rc + + continue + } + + key := sdkclient.ClientTupleKey{ + User: t.User, + Relation: t.Relation, + Object: t.Object, + } + if t.Condition != "" { + rc := &openfga.RelationshipCondition{Name: t.Condition} + if len(t.Context) > 0 { + rc.Context = &t.Context } - writes = append(writes, key) + key.Condition = rc } + writes = append(writes, key) } return writes, deletes