crates/sources: extend_from_catalog fixes

Use the _local_ scope, NOT the remote one, to determine the existing
local scope of a resource which is being kept or over-written.

We must also clear out imports of a local spec which is being replaced,
because the replacement is a fully inline model, and we're typically
going to indirect it before writing out resources, which would otherwise
result in duplicated imports.

Fixes #1590
Fixes #1558

crates/sources/src/merge.rs

@@ -217,10 +217,16 @@ where
                         entity,
                         catalog_name.as_ref(),
                         &root,
-                        Some(rhs.scope()),
+                        Some(lhs.scope()),
                     );
 
                     if let Some(last) = chain.last() {
+                        // We're fully replacing the lhs resource with the already-inlined
+                        // `model`, which means that all of its imports are no longer used.
+                        draft
+                            .imports
+                            .retain(|r| !r.scope.as_str().starts_with(lhs.scope().as_str()));
+
                         add_imports(draft, &chain);
                         *count += 1;
 

crates/sources/src/merge_test_src.yaml

@@ -1,4 +1,8 @@
 test://example/catalog.yaml:
+  import:
+    - flow://control/some/other/resource.yaml
+
+flow://control/some/other/resource.yaml:
   captures:
     acmeCo/one/new:
       endpoint:

crates/sources/src/merge_test_tgt.yaml

@@ -6,13 +6,12 @@ test://example/catalog.yaml:
       endpoint:
         connector:
           image: an/image
-          config: [old]
+          config: test://example/path/to/old/config
       bindings: []
 
   collections:
     acmeCo/collections/exists:
-      schema:
-        const: old-value
+      schema: test://example/path/to/old/schema
       key: [/int]
 
   materializations:
@@ -27,3 +26,7 @@ test://example/catalog.yaml:
     acmeCo/tests/exists: []
 
 test://example/acmeCo/flow.yaml: {}
+
+test://example/path/to/old/config: [old]
+
+test://example/path/to/old/schema: { const: old-value }

crates/sources/src/snapshots/sources__merge__test__merging.snap

@@ -114,6 +114,14 @@ DraftCatalog {
             depth: 2,
             resource: test://example/acmeCo/flow.yaml,
         },
+        Fetch {
+            depth: 2,
+            resource: test://example/path/to/old/config,
+        },
+        Fetch {
+            depth: 2,
+            resource: test://example/path/to/old/schema,
+        },
     ],
     imports: [
         Import {
@@ -158,5 +166,17 @@ DraftCatalog {
             content: ".. binary ..",
             content_dom: {"import":["acmeCo/flow.yaml"],"captures":{"acmeCo/captures/exists":{"endpoint":{"connector":{"image":"an/image","config":["updated"]}},"bindings":[],"expectPubId":"8866775533009900"}},"collections":{"acmeCo/collections/exists":{"schema":{"const":["updated-value"]},"key":["/int"],"expectPubId":"1100220033004400"}},"materializations":{"acmeCo/materializations/exists":{"endpoint":{"connector":{"image":"an/image","config":["also-updated"]}},"bindings":[]}},"tests":{"acmeCo/tests/exists":{"steps":[]}}},
         },
+        Resource {
+            resource: test://example/path/to/old/config,
+            content_type: "CONFIG",
+            content: ".. binary ..",
+            content_dom: "[\"old\"]",
+        },
+        Resource {
+            resource: test://example/path/to/old/schema,
+            content_type: "JSON_SCHEMA",
+            content: ".. binary ..",
+            content_dom: {"const":"old-value"},
+        },
     ],
 }