feat: a source can name linked assets by MRN and ingestion stores their IDs

internal/core/asset/export_test.go

@@ -0,0 +1,8 @@
+package asset
+
+// The tests that need a database live in asset_test: pgtest reaches this package through the
+// plugin registry, so importing it from here would close a cycle.
+var (
+	MustLoadProfile = mustLoadProfile
+	ValidCreate     = validCreate
+)

internal/core/asset/links.go

@@ -2,6 +2,7 @@ package asset
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"slices"
 	"strings"
@@ -61,6 +62,81 @@ func LinkIDs(value any) []string {
 	return nil
 }
 
+// mrnPrefix marks a link written by something that only knows the target's MRN, such as a
+// connector: IDs are minted here, so a source cannot name them.
+const mrnPrefix = "mrn://"
+
+// HasLinkMRNs reports whether metadata holds an MRN anywhere a link could be.
+func HasLinkMRNs(value any) bool {
+	switch v := value.(type) {
+	case string:
+		return strings.HasPrefix(v, mrnPrefix)
+	case []any:
+		return slices.ContainsFunc(v, HasLinkMRNs)
+	case []string:
+		return slices.ContainsFunc(v, func(item string) bool { return strings.HasPrefix(item, mrnPrefix) })
+	case map[string]any:
+		for _, item := range v {
+			if HasLinkMRNs(item) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// resolveLinkMRNs replaces, in the asset-control fields, each MRN with the ID of the asset it
+// names, on a copy of the metadata. An MRN that names nothing is left out; with deferred, its whole field is, so that a run
+// which creates both ends can fill the field in one go on its second pass instead of leaving a
+// partial value that would then count as curated.
+func (s *service) resolveLinkMRNs(ctx context.Context, asset *Asset, deferred bool) error {
+	if !HasLinkMRNs(asset.Metadata) {
+		return nil
+	}
+	// The caller keeps its map as it sent it: a run sends the same metadata again on its second pass.
+	metadata, err := cloneMetadata(asset.Metadata)
+	if err != nil {
+		return err
+	}
+	asset.Metadata = metadata
+	for _, f := range LinkFields(s.registry()) {
+		value, present := metamodel.ValueAt(metadata, f.Storage)
+		if !present || !HasLinkMRNs(value) {
+			continue
+		}
+		raw := LinkIDs(value)
+		ids := make([]any, 0, len(raw))
+		missing := false
+		for _, item := range raw {
+			if !strings.HasPrefix(item, mrnPrefix) {
+				ids = append(ids, item)
+				continue
+			}
+			target, err := s.repo.GetByMRN(ctx, item)
+			if errors.Is(err, ErrNotFound) || errors.Is(err, ErrAssetNotFound) {
+				missing = true
+				continue
+			}
+			if err != nil {
+				return fmt.Errorf("resolving linked asset %s: %w", item, err)
+			}
+			ids = append(ids, target.ID)
+		}
+		var resolved any = ids
+		if _, single := value.(string); single && len(ids) > 0 {
+			resolved = ids[0]
+		}
+		if len(ids) == 0 || (missing && deferred) {
+			resolved = nil
+		}
+		parts := strings.Split(strings.TrimPrefix(f.Storage, "metadata."), ".")
+		if err := setMetadataValue(metadata, parts, resolved); err != nil {
+			return fmt.Errorf("field %q: %w", f.ID, err)
+		}
+	}
+	return nil
+}
+
 // checkLinks rejects asset-control values naming the asset itself, or adding an asset that does
 // not exist or whose asset_type the field's targetAssetTypes leaves out. Links the asset already had are kept even once their target is deleted, so an
 // unrelated edit or a re-ingest never fails on them.

internal/core/asset/links_test.go

@@ -145,3 +145,41 @@ fields:
 		}
 	}
 }
+
+func TestAnMRNLinkResolvesOnceItsTargetExists(t *testing.T) {
+	svc := newLinkService(t)
+	ctx := context.Background()
+	first := validCreate("source")
+	mrn := "mrn://table/test/source"
+	first.MRN = &mrn
+	first.Metadata["example"] = map[string]any{"applies_to": []any{"mrn://table/test/later"}}
+	first.DeferLinks = true
+	source, err := svc.Create(ctx, first)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if value, present := metamodel.ValueAt(source.Metadata, "metadata.example.applies_to"); present {
+		t.Fatalf("a deferred field with an MRN that names nothing yet is not stored: %v", value)
+	}
+	later := validCreate("later")
+	laterMRN := "mrn://table/test/later"
+	later.MRN = &laterMRN
+	target, err := svc.Create(ctx, later)
+	if err != nil {
+		t.Fatal(err)
+	}
+	updated, err := svc.Update(ctx, source.ID, UpdateInput{
+		Metadata: map[string]any{"example": map[string]any{"applies_to": []any{"mrn://table/test/later"}}},
+		FromSync: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	value, _ := metamodel.ValueAt(updated.Metadata, "metadata.example.applies_to")
+	if got := LinkIDs(value); len(got) != 1 || got[0] != target.ID {
+		t.Fatalf("the link must hold the target's ID once it exists: %v", got)
+	}
+	if !HasLinkMRNs(map[string]any{"a": []any{"x", "mrn://t/p/n"}}) || HasLinkMRNs(map[string]any{"a": "x"}) {
+		t.Fatal("HasLinkMRNs finds an MRN anywhere in the metadata")
+	}
+}

internal/core/asset/metamodel_postgres_test.go

@@ -1,11 +1,12 @@
-package asset
+package asset_test
 
 import (
 	"context"
 	"errors"
 	"testing"
 	"time"
 
+	"github.com/marmotdata/marmot/internal/core/asset"
 	"github.com/marmotdata/marmot/internal/core/metamodel"
 	"github.com/marmotdata/marmot/internal/metrics"
 	"github.com/marmotdata/marmot/internal/store/postgres/pgtest"
@@ -18,9 +19,9 @@ func (dbRecorder) RecordDBQuery(context.Context, string, time.Duration, bool) {}
 func TestMetamodelPostgresWrites(t *testing.T) {
 	pool := pgtest.TempDB(t)
 	ctx := context.Background()
-	repo := NewPostgresRepository(pool, dbRecorder{})
-	svc := NewService(repo, WithMetamodel(mustLoadProfile(t)))
-	input := validCreate("postgres-metamodel")
+	repo := asset.NewPostgresRepository(pool, dbRecorder{})
+	svc := asset.NewService(repo, asset.WithMetamodel(asset.MustLoadProfile(t)))
+	input := asset.ValidCreate("postgres-metamodel")
 	if err := pool.QueryRow(ctx, "SELECT id FROM users WHERE username = 'admin'").Scan(&input.CreatedBy); err != nil {
 		t.Fatal(err)
 	}
@@ -46,7 +47,7 @@ func TestMetamodelPostgresWrites(t *testing.T) {
 	if _, err := svc.PatchFields(ctx, created.ID, 2, map[string]any{"retention": -1}); err == nil {
 		t.Fatal("invalid value persisted")
 	}
-	if _, err := svc.PatchFields(ctx, created.ID, 1, map[string]any{"retention": 1}); !errors.Is(err, ErrVersionConflict) {
+	if _, err := svc.PatchFields(ctx, created.ID, 1, map[string]any{"retention": 1}); !errors.Is(err, asset.ErrVersionConflict) {
 		t.Fatalf("stale patch: %v", err)
 	}
 
@@ -57,7 +58,7 @@ func TestMetamodelPostgresWrites(t *testing.T) {
 	}
 	start := make(chan struct{})
 	results := make(chan error, 2)
-	for _, a := range []*Asset{stored, other} {
+	for _, a := range []*asset.Asset{stored, other} {
 		go func() { <-start; results <- repo.Update(ctx, a) }()
 	}
 	close(start)
@@ -67,7 +68,7 @@ func TestMetamodelPostgresWrites(t *testing.T) {
 		switch {
 		case err == nil:
 			successes++
-		case errors.Is(err, ErrVersionConflict):
+		case errors.Is(err, asset.ErrVersionConflict):
 			conflicts++
 		default:
 			t.Fatal(err)

internal/core/asset/service.go

@@ -68,15 +68,18 @@ type Environment struct {
 } // @name Environment
 
 type CreateInput struct {
-	Name          *string                `json:"name" validate:"required"`
-	MRN           *string                `json:"mrn" validate:"required"`
-	Type          string                 `json:"type" validate:"required"`
-	Providers     []string               `json:"providers" validate:"required"`
-	Description   *string                `json:"description"`
-	Metadata      map[string]interface{} `json:"metadata"`
-	Schema        map[string]string      `json:"schema"`
-	Tags          []string               `json:"tags"`
-	CreatedBy     string                 `json:"created_by" validate:"required"`
+	Name        *string                `json:"name" validate:"required"`
+	MRN         *string                `json:"mrn" validate:"required"`
+	Type        string                 `json:"type" validate:"required"`
+	Providers   []string               `json:"providers" validate:"required"`
+	Description *string                `json:"description"`
+	Metadata    map[string]interface{} `json:"metadata"`
+	Schema      map[string]string      `json:"schema"`
+	Tags        []string               `json:"tags"`
+	CreatedBy   string                 `json:"created_by" validate:"required"`
+	// DeferLinks is set by a discovery run on its first pass: a link field naming, by MRN, an asset
+	// that does not exist yet is left out whole, for the run to send again once its batch is in.
+	DeferLinks    bool                   `json:"-"`
 	Sources       []AssetSource          `json:"sources"`
 	Environments  map[string]Environment `json:"environments"`
 	ExternalLinks []ExternalLink         `json:"external_links"`
@@ -109,6 +112,8 @@ type UpdateInput struct {
 	// FromSync marks a discovery run: the source fills governed fields that are still
 	// empty but never overwrites a value already set, so it needs no expected version.
 	FromSync bool `json:"-"`
+	// DeferLinks has the meaning it has in CreateInput.
+	DeferLinks bool `json:"-"`
 }
 
 type Filter struct {
@@ -498,6 +503,9 @@ func (s *service) Create(ctx context.Context, input CreateInput) (*Asset, error)
 	if err := s.validateAsset(asset); err != nil {
 		return nil, err
 	}
+	if err := s.resolveLinkMRNs(ctx, asset, input.DeferLinks); err != nil {
+		return nil, err
+	}
 	if err := s.checkLinks(ctx, asset.ID, asset.Metadata, nil); err != nil {
 		return nil, err
 	}
@@ -700,6 +708,9 @@ func (s *service) Update(ctx context.Context, id string, input UpdateInput) (*As
 			return nil, err
 		}
 	}
+	if err := s.resolveLinkMRNs(ctx, asset, input.DeferLinks); err != nil {
+		return nil, err
+	}
 	if err := s.checkLinks(ctx, asset.ID, asset.Metadata, oldAsset.Metadata); err != nil {
 		return nil, err
 	}

internal/core/asset/terms_pg_test.go

@@ -1,10 +1,11 @@
-package asset
+package asset_test
 
 import (
 	"context"
 	"testing"
 	"time"
 
+	"github.com/marmotdata/marmot/internal/core/asset"
 	"github.com/marmotdata/marmot/internal/metrics"
 	"github.com/marmotdata/marmot/internal/store/postgres/pgtest"
 )
@@ -31,7 +32,7 @@ func TestAssetsByTermListsTheTermAssets(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	assets, total, err := NewPostgresRepository(pool, noopRecorder{}).GetAssetsByTerm(ctx, termID, 10, 0)
+	assets, total, err := asset.NewPostgresRepository(pool, noopRecorder{}).GetAssetsByTerm(ctx, termID, 10, 0)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/core/runs/link_mrns_test.go

@@ -0,0 +1,52 @@
+package runs
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/marmotdata/marmot/internal/core/asset"
+	"github.com/marmotdata/marmot/internal/core/lineage"
+	"github.com/marmotdata/marmot/internal/core/metamodel"
+	"github.com/marmotdata/marmot/internal/store/postgres/pgtest"
+	"github.com/stretchr/testify/require"
+)
+
+const linkProfile = `formatVersion: 1
+id: example
+version: 1
+defaultLocale: en
+fields:
+  - id: parent_area
+    type: list
+    itemType: string
+    core: true
+    nullable: true
+    storage: metadata.example.parent_area
+    presentation:
+      labelKey: example.parent_area.label
+      control: asset
+`
+
+func TestProcessEntities_LinksByMRNAnAssetCreatedLaterInTheSameRun(t *testing.T) {
+	pool := pgtest.TempDB(t)
+	ctx := context.Background()
+	registry, err := metamodel.Load(strings.NewReader(linkProfile))
+	require.NoError(t, err)
+	assets := asset.NewService(asset.NewPostgresRepository(pool, recorderStub{}), asset.WithMetamodel(registry))
+	service := NewService(NewPostgresRepository(pool), assets, lineage.NewService(lineage.NewPostgresRepository(pool), assets), nil, recorderStub{})
+
+	child := tableInput("ventas")
+	child.Metadata = map[string]interface{}{
+		"example": map[string]interface{}{"parent_area": []interface{}{"mrn://table/postgresql/corporacion", "mrn://table/postgresql/nowhere"}},
+	}
+	// The child comes first: its parent does not exist when it is created.
+	process(t, service, "areas", []CreateAssetInput{child, tableInput("corporacion")}, nil)
+
+	parent, err := assets.GetByMRN(ctx, "mrn://table/postgresql/corporacion")
+	require.NoError(t, err)
+	stored, err := assets.GetByMRN(ctx, "mrn://table/postgresql/ventas")
+	require.NoError(t, err)
+	value, _ := metamodel.ValueAt(stored.Metadata, "metadata.example.parent_area")
+	require.Equal(t, []string{parent.ID}, asset.LinkIDs(value), "the MRN becomes the parent's ID; one that names nothing is dropped")
+}

internal/core/runs/service.go

@@ -174,6 +174,12 @@ type GlossarySyncer interface {
 	SyncTerms(ctx context.Context, source string, inputs []glossary.TermInput) (*glossary.SyncResult, error)
 }
 
+// linkedAsset is an asset whose metadata names others by MRN.
+type linkedAsset struct {
+	id, mrn  string
+	metadata map[string]interface{}
+}
+
 type service struct {
 	repo               Repository
 	assetService       asset.Service
@@ -291,6 +297,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr
 		return nil, fmt.Errorf("getting last run checkpoints: %w", err)
 	}
 
+	var relink []linkedAsset
 	response := &ProcessAssetsResponse{
 		Assets:        make([]AssetResult, 0, len(assets)),
 		Lineage:       make([]LineageResult, 0, len(lineage)),
@@ -350,6 +357,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr
 				QueryLanguage: ast.QueryLanguage,
 				Sources:       ast.Sources,
 				CreatedBy:     run.CreatedBy,
+				DeferLinks:    true,
 			}
 			created, err := s.assetService.Create(ctx, createInput)
 			if created != nil {
@@ -388,6 +396,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr
 				Sources:          ast.Sources,
 				SkipNotification: true,
 				FromSync:         true,
+				DeferLinks:       true,
 			}
 			// Not cached when another writer created the asset after the fetch above.
 			existingAsset := existingAssets[assetMRN]
@@ -425,6 +434,10 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr
 			}
 		}
 
+		if status != StatusFailed && assetID != "" && asset.HasLinkMRNs(ast.Metadata) {
+			relink = append(relink, linkedAsset{id: assetID, mrn: assetMRN, metadata: ast.Metadata})
+		}
+
 		result := AssetResult{
 			Name:     ast.Name,
 			Type:     ast.Type,
@@ -454,6 +467,17 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr
 		}
 	}
 
+	// A link by MRN may name an asset this same batch created after the one that holds it, so the
+	// fields left out on the way in are written now that every target exists.
+	// ponytail: an asset the run found unchanged is not revisited; a target that another source
+	// adds later is linked the next time this asset changes.
+	for _, linked := range relink {
+		input := asset.UpdateInput{Metadata: linked.metadata, SkipNotification: true, FromSync: true}
+		if _, err := s.assetService.Update(ctx, linked.id, input); err != nil {
+			log.Warn().Err(err).Str("asset_mrn", linked.mrn).Msg("Failed to link assets by MRN")
+		}
+	}
+
 	staleEntities := s.GetStaleEntities(ctx, lastCheckpoints, currentMRNs)
 	for _, staleMRN := range staleEntities {
 		if err := s.assetService.DeleteByMRN(ctx, staleMRN); err != nil {