ILLDEV-484 Add directory import endpoint

[JanisSaldabols]

No description provided.

[Copilot]

🟡 Changes recommended

Conflict-policy races, validation gaps, and incompatible name constraints can produce incorrect imports and API failures.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds an authenticated NDJSON endpoint for importing complete directory aggregates with per-record conflict handling and validation.

Changes:

• Adds streaming entry, tier, and network imports.
• Adds persistence, validation, authorization, and size limits.
• Introduces business-key constraints and integration coverage.

File summaries

File	Description
directory/test/import_test.go	Adds end-to-end import tests.
directory/sqlc.yaml	Preserves nullable generated name types.
directory/query.sql	Adds import locking and update queries.
directory/oapi-codegen.yaml	Retains extension-referenced schemas.
directory/migrations/006_import_business_keys.up.sql	Adds import business-key constraints.
directory/migrations/006_import_business_keys.down.sql	Reverts those constraints.
directory/import/service/importer.go	Implements NDJSON stream processing.
directory/import/service/importer_test.go	Tests importer behavior and limits.
directory/import/service/decode.go	Strictly decodes import records.
directory/import/model/models.go	Defines and validates import aggregates.
directory/import/model/models_test.go	Tests aggregate validation.
directory/import/model/config.go	Models imported configuration data.
directory/import/db/tier_network.go	Persists tiers, networks, and memberships.
directory/import/db/repo.go	Provides transaction and lookup helpers.
directory/import/db/repo_test.go	Adds database integration tests.
directory/import/db/entry.go	Persists complete entry aggregates.
directory/enhancedcontext/enhancedcontext_test.go	Fixes test indentation.
directory/domain/hierarchy.go	Centralizes hierarchy validation.
directory/domain/hierarchy_test.go	Tests hierarchy rules.
directory/descriptors/ModuleDescriptor-template.json	Registers the import route.
directory/app/import_limit.go	Adds authorization and body limiting.
directory/app/import_limit_test.go	Tests middleware behavior.
directory/app/app.go	Wires importer and middleware.
directory/api/import.go	Implements the HTTP import operation.
directory/api/import_test.go	Tests endpoint response handling.
directory/api/import_contract_test.go	Verifies the OpenAPI contract.
directory/api/impl.go	Injects the importer service.
directory/api/entries.go	Reuses shared hierarchy validation.
directory/api.yaml	Defines the endpoint and record schemas.

Review details

Suppressed comments (2)

directory/query.sql:492

• As with the tier lookup, this row lock provides no serialization when the network does not yet exist. Concurrent skip/update imports can race at the unique insert and report a failed record rather than honoring the conflict policy. Make creation/conflict detection atomic or acquire a business-key advisory lock before lookup.

-- name: LockNetworkByBusinessKey :one
SELECT * FROM networks
WHERE consortium = @consortium AND name = @name
FOR UPDATE;

directory/import/db/entry.go:49

• This path permits an existing consortium to be changed to another type when it has no entry children, but any tiers or networks owned by that entry remain linked to it. Those aggregates then violate the repository invariant enforced by resolveConsortium and can no longer be imported by business key. Reject the demotion while dependent aggregates exist, or remove/reassign them transactionally.

	if aggregate.Data.Type == "Consortium" || (exists && existing.Type == "Consortium") {
		if err := queries.LockConsortiumEntryChanges(ctx); err != nil {
			return model.RepoResult{}, fmt.Errorf("lock consortium entry changes")
		}
	}
	if aggregate.Data.Type == "Consortium" && (!exists || existing.Type != "Consortium") {

• Files reviewed: 28/29 changed files
• Comments generated: 4
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟡 Changes recommended

Unresolved critical and moderate review findings remain.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

directory/api.yaml:11

• The repository resolves every tier/network member immediately, so a member entry that appears later in the NDJSON stream makes the aggregate fail permanently. The documented requirement only says that parent entries precede children, so a stream satisfying that rule can still lose a tier or network. Either defer these records until their referenced entries exist or document that all referenced entries must precede tier/network records.

      description: Imports ordered NDJSON entry, tier, and network aggregates. Parent entries must appear in earlier committed records.

directory/app/import_limit.go:27

• This wraps http.NoBody for authorized empty POSTs. The handler's empty-body check (request.Body == http.NoBody) will therefore no longer match; the importer reads EOF and returns a successful empty result, violating the required request body contract. Preserve the http.NoBody case before installing MaxBytesReader (or otherwise reject it here).

		request.Body = http.MaxBytesReader(writer, request.Body, maxBytes)
		next.ServeHTTP(writer, request)

directory/import/service/importer.go:103

• Repository cancellation and deadline errors are treated as ordinary per-record failures here, so the importer continues reading and processing every remaining record. With the 2 GiB request limit, a timed-out or disconnected request can still drain a very large body and generate one failed database call per record instead of terminating promptly; propagate context.Canceled/context.DeadlineExceeded as a fatal import error and check the context while scanning.

	if err != nil {
		incrementFailed(result, record.recordType)
		appendError(result, line, record.recordType, record.key, err.Error())
		return

• Files reviewed: 35/37 changed files
• Comments generated: 5
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Secondary-symbol locking and fatal propagation of cancellation or deadline errors must be addressed before approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

• Files reviewed: 36/38 changed files
• Comments generated: 2
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Resolve the critical lock-ordering deadlock and the two moderate concurrency and identity-key issues.

Get a fresh assessment by requesting another Copilot review.

Review details

Suppressed comments (2)

directory/import/db/entry.go:390

• The advisory lock here is shared only by import transactions; AddEntry inserts symbols without taking LockEntryImportKey. If that API creates this symbol after the import's initial lookup but before UpsertSymbol, the unique (authority, symbol) insert fails and the record is reported as failed, even for skip or update. Use the same key lock in the API path, or retry this unique-conflict case and reapply the selected policy.

	for _, symbol := range data.Symbols {
		if _, err := queries.UpsertSymbol(ctx, db.UpsertSymbolParams{Owner: entryID, Authority: symbol.Authority, Symbol: symbol.Symbol}); err != nil {
			return err

directory/import/model/models.go:77

• String() is used as an identity key for duplicate detection and lock revalidation, but authority + ":" + symbol is ambiguous while the import schema accepts any non-empty strings. Distinct database symbols such as (A:B,C) and (A,B:C) collide here, so valid records can be rejected or mappings can be skipped/retried. Use a comparable struct key for identity, or reject : consistently in the schema and validation.

func (s SymbolRef) String() string { return s.Authority + ":" + s.Symbol }

• Files reviewed: 36/38 changed files
• Comments generated: 1
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

🔵 Needs a closer look

Five moderate error-propagation and lock-order issues remain unresolved.

Review details

Suppressed comments (5)

directory/import/db/entry.go:336

• lockEntryRows acquires all involved entry rows in UUID order, but the existing /entries update path locks its target row first and then its parent (directory/api/entries.go:856-888). An import of child A under parent B can therefore hold A while waiting for B, while a concurrent update of B with parent A holds B while waiting for A, producing a PostgreSQL deadlock (and a 500 on the non-retrying API path). Use one shared deterministic lock order for both paths, or otherwise coordinate these hierarchy locks.

func lockEntryRows(ctx context.Context, queries *db.Queries, ids ...uuid.UUID) (map[uuid.UUID]db.Entry, error) {
	entries := make(map[uuid.UUID]db.Entry, len(ids))
	for _, id := range orderedUniqueEntryIDs(ids...) {
		entry, err := queries.EntryByIdForUpdate(ctx, id)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errImportEntryMappingChanged
		}

directory/import/db/tier_network.go:46

• When the business-key lookup fails for a reason other than pgx.ErrNoRows, this drops lookupErr. The importer then exposes only resolve tier ..., so connection/transaction failures lose their cause and cannot be diagnosed with errors.Is/As; wrap the lookup error.

		return model.RepoResult{}, fmt.Errorf("resolve tier %s", key)

directory/import/db/tier_network.go:116

• When the business-key lookup fails for a reason other than pgx.ErrNoRows, this drops lookupErr. The importer then exposes only resolve network ..., so connection/transaction failures lose their cause and cannot be diagnosed with errors.Is/As; wrap the lookup error.

		return model.RepoResult{}, fmt.Errorf("resolve network %s", key)

directory/import/db/tier_network.go:75

• A commit failure is returned without its underlying error, unlike the entry importer. This makes a failed tier commit appear only as commit tier ... import in the API's item error and hides whether the cause was cancellation, a connection failure, or a database conflict; preserve err with %w.

		return model.RepoResult{}, fmt.Errorf("commit tier %s import", key)

directory/import/db/tier_network.go:145

• A commit failure is returned without its underlying error, unlike the entry importer. This makes a failed network commit appear only as commit network ... import in the API's item error and hides whether the cause was cancellation, a connection failure, or a database conflict; preserve err with %w.

		return model.RepoResult{}, fmt.Errorf("commit network %s import", key)

• Files reviewed: 36/38 changed files
• Comments generated: 0 new
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

🟡 Changes recommended

Critical and moderate correctness, concurrency, and error-handling issues remain unresolved.

Get a fresh assessment by requesting another Copilot review.

Review details

Suppressed comments (5)

directory/import/db/entry.go:345

• This UUID-sorted row-lock order conflicts with the existing UpdateEntry path, which locks the target row first and then its parent (directory/api/entries.go:856-888). When the parent UUID sorts before the target, an import can hold the parent while waiting for the target as the API update holds the target while waiting for the parent, producing a PostgreSQL deadlock. Use a lock order shared by both paths, or coordinate/retry the API transaction as well.

	for _, id := range orderedUniqueEntryIDs(ids...) {
		entry, err := queries.EntryByIdForUpdate(ctx, id)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errImportEntryMappingChanged
		}
		if err != nil {
			return nil, err

directory/import/db/tier_network.go:219

• These assignment-write failures discard the underlying database error, so the import result only reports replace tier assignments and loses useful PostgreSQL codes/causes for diagnosis and retry decisions. Preserve the cause with %w, as the business-key and commit paths in this PR already do.

		return fmt.Errorf("replace tier assignments")
	}
	for _, entry := range entries {
		if _, err := queries.CreateEntryTier(ctx, db.CreateEntryTierParams{Entry: entry.ID, Tier: tierID}); err != nil {
			return fmt.Errorf("replace tier assignments")

directory/import/db/tier_network.go:234

• As in the tier path, these errors are replaced by a generic replace network assignments string, hiding the database cause and SQLSTATE from the per-record import diagnostics. Wrap both failures with %w so callers can retain the original error.

		return fmt.Errorf("replace network assignments")
	}
	for index, entry := range entries {
		if _, err := queries.CreateEntryNetwork(ctx, db.CreateEntryNetworkParams{Entry: entry.ID, Network: networkID, Priority: assignments[index].Priority}); err != nil {
			return fmt.Errorf("replace network assignments")

directory/import/db/tier_network.go:66

• The consortium row is locked before this insert for import-vs-import serialization, but AddTier does not take that lock. If the API creates the same (consortium,name) after LockTierByBusinessKey observes no row, this insert gets the new unique-constraint violation; ImportTier returns it as a failed record instead of applying skip or update. Coordinate the API and importer lock/retry paths for this race.

		var created db.Tier
		created, err = queries.CreateTier(ctx, db.CreateTierParams{Name: name, Consortium: consortium.ID, Level: aggregate.Data.Level, Type: aggregate.Data.Type, Cost: aggregate.Data.Cost})
		tierID = created.ID

directory/import/db/tier_network.go:136

• The consortium row is locked before this insert for import-vs-import serialization, but AddNetwork does not take that lock. If the API creates the same (consortium,name) after LockNetworkByBusinessKey observes no row, this insert gets the new unique-constraint violation; ImportNetwork returns it as a failed record instead of applying skip or update. Coordinate the API and importer lock/retry paths for this race.

		var created db.Network
		created, err = queries.CreateNetwork(ctx, db.CreateNetworkParams{Name: name, Consortium: consortium.ID, Reciprocal: aggregate.Data.Reciprocal})
		networkID = created.ID

• Files reviewed: 36/38 changed files
• Comments generated: 2
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

🔵 Needs a closer look

Three moderate issues remain in conflict ordering, unique-key race handling, and advisory-lock key construction.

Review details

Suppressed comments (4)

Previously missed (1) — in code that hasn't changed since the last review.

directory/import/db/tier_network.go:48

• Assignment resolution happens before the business-key conflict check. If this tier already exists and the record uses conflictPolicy=skip (or fail) but contains a missing assignment entry, resolveAndLockAssignments returns an error and the record is counted as failed instead of being skipped (or reported as an existing-tier conflict). Resolve/lock the consortium and tier key first, and only resolve assignments for a create/update path.

This issue also appears on line 123 of the same file.

directory/import/db/tier_network.go:125

• This has the same conflict-ordering problem for networks: a missing assignment is resolved before LockNetworkByBusinessKey, so an existing network with conflictPolicy=skip is reported as failed rather than skipped. Check the network business key before resolving assignment members, while retaining assignment resolution for create/update.

	consortium, assignments, err := resolveAndLockAssignments(ctx, queries, aggregate.Key.Consortium, refs)
	if err != nil {
		return model.RepoResult{}, err

directory/import/db/tier_network.go:225

• A tier/network import can race with the regular AddTier/AddNetwork endpoint: the business-key lookup may see no row, then the import INSERT loses the unique-key race with SQLSTATE 23505. This retry helper handles lock/deadlock errors but not tiers_consortium_name_unique or networks_consortium_name_unique, so skip and update can be returned as failed instead of being re-evaluated after the concurrent commit; retry those constraint violations.

	return pgErr.Code == "40P01" || pgErr.Code == "40001" || pgErr.Code == "55P03"

directory/query.sql:29

• This advisory key uses : as a separator, but NormalizeAndValidate deliberately permits : in authorities and symbols. Consequently authority=A:B, symbol=C and authority=A, symbol=B:C acquire the same lock; two aggregates containing multiple such aliases can sort those aliases in opposite orders, deadlock, and consume the five-attempt retry budget. Encode the two components unambiguously (for example with length prefixes) before hashing.

• Files reviewed: 38/40 changed files
• Comments generated: 0 new
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

Note

Copilot was unable to run its full agentic suite in this review.

Pull request overview

Copilot reviewed 38 out of 40 changed files in this pull request and generated 3 comments.

Suppressed comments (3)

directory/test/concurrency_test.go:1

• The tracer matches on -- name: ... comment markers, but SQLC query name comments are not sent to PostgreSQL and typically are not present in data.SQL. This makes the pause logic fail to trigger (and can make the test behavior unreliable). Match against the actual SQL text (e.g., FOR UPDATE vs FOR UPDATE NOWAIT) or another stable discriminator instead of -- name:.
  directory/test/concurrency_test.go:1
• pg_stat_activity.query will contain the SQL statement text executed by PostgreSQL, not SQLC's -- name: metadata. As written, this LIKE '%EntryByIdForUpdate%' predicate will never match and the require.Eventually will time out. Switch the predicates to match real SQL fragments (e.g., LIKE '%FOR UPDATE%' and NOT LIKE '%NOWAIT%'), or use an alternate approach (e.g., pg_locks correlation) that doesn't rely on SQLC query names.
  directory/test/concurrency_test.go:1
• lockUnavailableTracer is duplicated in multiple test packages (e.g., also appears in directory/import/db/repo_test.go). Consider factoring this into a shared test helper (or reusing one definition) to avoid drift if tracer logic changes.

[Copilot]

🔵 Needs a closer look

Resolve the tier/network business-key race handling so update and skip conflict policies behave correctly.

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

directory/import/db/tier_network.go:225

• The import's unique-key race handling does not include the new tiers_consortium_name_unique and networks_consortium_name_unique violations. AddTier/AddNetwork can insert the same business key without taking the import's entry lock, so if that API request wins, this returns a non-retryable 23505; conflictPolicy=update is then reported as failed instead of retrying to update the row (and skip is not counted as skipped). Treat these two constraint violations as retryable, or make the API use the same business-key lock.

• Files reviewed: 38/40 changed files
• Comments generated: 0 new
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

🟡 Changes recommended

The migration can abort on existing NULL or whitespace-only tier/network names; provide upgrade remediation before approval.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 38/40 changed files
• Comments generated: 1
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[Copilot]

🟡 Changes recommended

Unresolved critical persistence-error disclosure and moderate per-assignment database round-trip findings block approval.

Get a fresh assessment by requesting another Copilot review.

Review details

Suppressed comments (1)

directory/import/db/tier_network.go:199

• For a tier or network with N assignments, this loop performs one EntryBySymbol round trip per assignment, then lockAssignmentEntryRows and lockEntryMappings each issue another per-entry query. Since a record may be up to 1 MiB and has no assignment-count limit, a large export can hold a transaction open for thousands of sequential database round trips; resolve and lock the references with set-based/batched queries while preserving the sorted lock order.

	for _, ref := range refs {
		entry, err := queries.EntryBySymbol(ctx, db.EntryBySymbolParams{Authority: ref.Authority, Symbol: ref.Symbol})
		if errors.Is(err, pgx.ErrNoRows) {
			assignments = append(assignments, resolvedAssignment{ref: ref})
			mappings = append(mappings, entryMapping{ref: ref})
			continue
		}
		if err != nil {
			return db.Entry{}, nil, fmt.Errorf("resolve entry %s: %w", ref.String(), err)
		}
		assignments = append(assignments, resolvedAssignment{ref: ref, entry: &entry})
		entryIDs = append(entryIDs, entry.ID)
		mappings = append(mappings, entryMapping{ref: ref, expectedOwner: &entry.ID})

• Files reviewed: 40/42 changed files
• Comments generated: 1
• Review effort level: Lite (auto)

Note

Copilot is running an experiment and ran this review at Lite.

[jakub-id]

please add an example in the dierctory README that uses curl to import

[jakub-id]

re Copilot comment about internal DB errors -- I think this is fine for this particular endpoint because it's used by devops and they would like to see verbatim DB errors during migration