feat(accounts): the object-storage datasource provider is resolved from the composed service, not compiled into the registry

module/services/README.md

@@ -13,8 +13,10 @@ behind `auth-gateway`.
 (codefly-dev/service-object-storage). The host keeps no store client of its
 own: accounts reads every object-storage datasource through it, under a
 per-call binding of the source's bucket and credential, and the gateway owns
-the store protocol and the public-destination guard — see
-`accounts/AGENTS.md`.
+the store protocol and the public-destination guard. The object-storage
+datasource provider exists only where this service is composed: accounts
+registers it when the dependency resolves, and a composition that leaves it out
+offers no object-storage sources — see `accounts/AGENTS.md`.
 
 Run the complete local dependency graph from `module/`:
 

module/services/accounts/AGENTS.md

@@ -1210,6 +1210,42 @@ prefix leaves the filter where it is.
   envelope connector applies it to the service's listing before the version is
   digested (#1211).
 
+## A datasource provider is compiled in, or resolved from a composed service
+
+The connector registry (`pkg/business/datasource_connectors.go`) is built at
+start with the providers compiled into this host — `github` (conformant), `api`
+and `crawler` (registered non-conformant with their gap). **Object storage is
+not one of them.** It is the first provider resolved from a composed service:
+`datasource_objectstore.go` registers it only once the declared
+`object-storage` dependency resolves to a service that can read a tenant's
+bucket, and holds everything the host does for that provider, so the generic
+catalog, admission and sync paths name no object-storage type.
+
+The resolution reuses what the registry already has — nothing beside it
+decides what a provider is:
+
+- **Registration** goes through `Registry.Register`, so the descriptor is
+  validated and the interface checked exactly as for a compiled-in provider,
+  and the connector is wrapped in `connector.Budgeted`.
+- **The per-credential budget stays host-side.** It meters across every source
+  and replica (`datasource_credential_budgets`), so it is shared state with a
+  tenancy; the composed service never sees it.
+- **The credential is per call** (envelope clause 3). The host decrypts it from
+  the source on each call and puts it on that call's binding; the service
+  neither returns nor stores it.
+- **The conformance suite runs against the provider as registered**
+  (`TestObjectStorageAsRegisteredPassesTheConformanceSuite`): the connector the
+  registry holds once the provider is resolved, on sources connected through
+  `AddSource`, reading their credential and retained listings through the host.
+- **Absent means absent.** A composition that leaves the service out — or
+  `codefly run service … --exclude-dependency saas-starter/object-storage` —
+  has no `upload` entry in `GetDatasourceCatalog`, and a source of it is
+  refused by name, not as an unknown provider.
+
+The four provider constants (`github`, `api`, `crawler`, `upload`) stay until a
+provider contract is published; converting `github`, `api` and `crawler` to
+composed services is not done here.
+
 ## A datasource source is refreshed by three things, and one of them covers every provider
 
 Ranked by latency, a source's content is refreshed by a push (near-live), by a
@@ -1394,14 +1430,16 @@ keeps what is per-source state:
   `GetDatasourceCatalog` reports `accepts_new_sources` and `AddSource` takes a
   new source with its prefix and key filter. That is the whole admin path a
   composed consumer drives — catalog, `AddSource`, sync — with no portal step.
-- **Admission needs a service that can read a bucket.** At start `work.go`
-  wires the composed service through `Service.SetObjectStorage`, which first
-  asks it, unbound, whether it reads a batch (`objectstore.CheckService`) and
-  refuses to wire one that does not. A composition without the service, or with one
-  from before the binding (0.0.18 and earlier; this module pins 0.0.19),
-  leaves object storage unavailable: the catalog says so, `AddSource` refuses
-  with `ErrObjectStorageUnavailable`, and an existing source's sync fails by
-  name.
+- **The provider exists only where its service resolves.** Object storage is
+  not compiled into the registry (see the next section). At start `work.go`
+  resolves the declared `object-storage` dependency and hands it to
+  `Service.SetObjectStorage`, which first asks it, unbound, whether it reads a
+  batch (`objectstore.CheckService`) and only then registers the provider. A
+  composition without the service, or with one from before the binding (0.0.18
+  and earlier; this module pins 0.0.19), has no `upload` provider: the catalog
+  does not list it, `AddSource` refuses with `ErrObjectStorageUnavailable`, an
+  existing source's sync fails with the same error, and the source reports it
+  as its conformance gap.
 - **A sync never empties a source that has files** — a host-owned invariant,
   whatever any service version answers. A closing snapshot with no files
   tombstones every entry mirrored from the source, and a service that wrongly

module/services/accounts/code/pkg/business/datasource_connectors.go

@@ -7,21 +7,21 @@ import (
 
 	"accounts/pkg/datasource/connector"
 	"accounts/pkg/datasource/github"
-	"accounts/pkg/datasource/objectstore"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
 // The host's datasource connectors, on the envelope in pkg/datasource/connector.
-// Two are conformant, each passing the conformance suite beside it: GitHub
-// (pkg/datasource/github/connector_conformance_test.go) and object storage,
-// read through the composed object-storage service
-// (pkg/datasource/objectstore/connector_conformance_test.go). The generic API
-// and the crawler still run on their own engines, outside the envelope; they
-// are registered non-conformant with their gap, so their existing sources keep
-// running and new ones are refused until each passes the suite (the owner's
-// settled question 7).
+// Two are conformant: GitHub, compiled in and passing the conformance suite
+// beside it (pkg/datasource/github/connector_conformance_test.go), and object
+// storage, which is not compiled into this registry — it is registered only
+// when the composed object-storage service resolves, and passes the suite as
+// registered (datasource_objectstore.go). The generic API and the crawler
+// still run on their own engines, outside the envelope; they are registered
+// non-conformant with their gap, so their existing sources keep running and
+// new ones are refused until each passes the suite (the owner's settled
+// question 7).
 
 // Why each legacy provider is not on the envelope yet, per connector. What
 // remains for each, clause by clause, and what it already honours, is in
@@ -64,10 +64,12 @@ func (s *Service) newDatasourceConnectorRegistry() *connector.Registry {
 		CredentialModes: []connector.CredentialMode{connector.CredentialNone},
 		Gap:             gapCrawlerDatasource,
 	}))
-	// Object storage reads through the composed object-storage service and
-	// passes the suite over its wire (objectstore/connector_conformance_test.go);
-	// metered like GitHub, per source credential.
-	must(r.Register(connector.Budgeted(objectstore.NewFilesConnector(s.uploadBucketForSource, objectManifests{s.store}), datasourceScheduler{s: s})))
+	// Object storage is not compiled in: it is registered when its composed
+	// service resolves (datasource_objectstore.go), here only if that happened
+	// before this registry was built.
+	if s.newUploadBucket != nil {
+		must(s.registerObjectStorage(r))
+	}
 	return r
 }
 
@@ -124,16 +126,15 @@ func DatasourceConformance(registry *connector.Registry, provider string) (bool,
 		return false, "the datasource connector is not configured"
 	}
 	d, ok := registry.Descriptor(provider)
-	if !ok {
+	switch {
+	case !ok && provider == DatasourceProviderUpload:
+		return false, ErrObjectStorageUnavailable.Error()
+	case !ok:
 		return false, "no connector is registered for this provider"
 	}
 	return d.Conformant, d.Gap
 }
 
-// ErrObjectStorageUnavailable: no composed object-storage service here can read
-// a tenant's bucket, so an object-storage source is neither admitted nor synced.
-var ErrObjectStorageUnavailable = errors.New("object storage is not available here: no composed object-storage service serves a per-call binding with a batched read")
-
 // admitNewDatasource refuses a new source of a provider the registry does not
 // admit, or that nothing composed here can serve, before anything is
 // validated, sealed or stored. The catalog reports the same answer.
@@ -143,14 +144,13 @@ func (s *Service) admitNewDatasource(provider string) error {
 	}
 	err := s.datasourceConnectors.AdmitNewSource(provider)
 	switch {
-	case err == nil && provider == DatasourceProviderUpload && s.newUploadBucket == nil:
-		// Conformant, but nothing here can read a bucket: no object-storage
-		// service is composed, or the one composed predates the per-call
-		// binding and the batched read (work.go asks it at start). A source
-		// admitted now would fail every sync.
-		return status.Error(codes.FailedPrecondition, ErrObjectStorageUnavailable.Error())
 	case err == nil:
 		return nil
+	case errors.Is(err, connector.ErrUnknownConnector) && provider == DatasourceProviderUpload:
+		// Not registered: no object-storage service is composed, or the one
+		// composed predates the per-call binding and the batched read
+		// (SetObjectStorage). The refusal names why, not "unknown".
+		return status.Error(codes.FailedPrecondition, ErrObjectStorageUnavailable.Error())
 	case errors.Is(err, connector.ErrUnknownConnector):
 		return status.Errorf(codes.InvalidArgument, "unknown datasource provider %q", provider)
 	default:
@@ -166,55 +166,11 @@ func connectorSource(source *DatasourceSource) connector.Source {
 	case source.Provider == DatasourceProviderGitHub:
 		src.Config = github.SourceConfig{Repo: source.Repo, Branch: source.Branch, InScope: datasourceFileScope(source)}
 	case source.Provider == DatasourceProviderUpload && source.Upload != nil:
-		cfg := source.Upload
-		src.Config = objectstore.SourceConfig{
-			Endpoint: cfg.Endpoint, Region: cfg.Region, Bucket: cfg.Bucket, Prefix: cfg.Prefix,
-			Scope: objectstore.Scope{InScope: uploadKeyScope(cfg), MaxObjects: cfg.MaxObjects},
-		}
+		src.Config = uploadSourceConfig(source.Upload)
 	}
 	return src
 }
 
-// uploadBucketForSource resolves one object-storage source's Bucket, per call:
-// the host decrypts the source's secret access key and binds it to that call,
-// and the connector never sees it.
-func (s *Service) uploadBucketForSource(ctx context.Context, src connector.Source) (objectstore.Bucket, error) {
-	if s.newUploadBucket == nil {
-		return nil, ErrObjectStorageUnavailable
-	}
-	source, err := s.store.GetDatasourceSourceByID(ctx, src.ID)
-	if err != nil {
-		return nil, err
-	}
-	if source == nil || source.OrgID != src.OrgID || source.Provider != DatasourceProviderUpload || source.Upload == nil {
-		return nil, fmt.Errorf("datasource source %s is not an object-storage source of org %s", src.ID, src.OrgID)
-	}
-	secret, err := s.datasourceCipher.DecryptSecret(ctx, DatasourceConnectorSecretPurpose(source.ID), source.CredentialSecretRef)
-	if err != nil {
-		return nil, fmt.Errorf("decrypt secret access key: %w", err)
-	}
-	cfg, _ := connectorSource(source).Config.(objectstore.SourceConfig)
-	return s.newUploadBucket(cfg, source.Upload.AccessKeyID, secret), nil
-}
-
-// objectManifests retains an object-storage source's listings in the store,
-// inside the source's org transaction.
-type objectManifests struct{ store Store }
-
-func (m objectManifests) Put(ctx context.Context, src connector.Source, version string, items []objectstore.Item) error {
-	return m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error {
-		return m.store.PutDatasourceObjectManifest(ctx, src.OrgID, src.ID, version, items)
-	})
-}
-
-func (m objectManifests) Get(ctx context.Context, src connector.Source, version string) (items []objectstore.Item, ok bool, err error) {
-	err = m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error {
-		items, ok, err = m.store.GetDatasourceObjectManifest(ctx, src.OrgID, src.ID, version)
-		return err
-	})
-	return items, ok, err
-}
-
 // githubRemoteForSource resolves the credentialed GitHub client for one source,
 // per call: the host holds the credential and the connector never sees it.
 func (s *Service) githubRemoteForSource(ctx context.Context, src connector.Source) (github.Remote, error) {

module/services/accounts/code/pkg/business/datasource_connectors_test.go

@@ -30,6 +30,8 @@ var suiteCall = regexp.MustCompile(`connectortest\.Run[A-Z][A-Za-z]*\(t,`)
 func TestEveryConformantConnectorRunsTheSuite(t *testing.T) {
 	svc, _ := business.NewService(nil)
 	svc.SetDatasourceConnector(nil, nil, "")
+	// Object storage is registered once its composed service resolves.
+	svc.SetDatasourceUploadBucketFactory(func(objectstore.SourceConfig, string, string) objectstore.Bucket { return nil })
 	conformant := svc.DatasourceConnectors().Conformant()
 	if len(conformant) < 2 {
 		t.Fatalf("the host registers %d conformant connectors, want GitHub and object storage", len(conformant))
@@ -108,25 +110,26 @@ func TestLegacyProvidersAreRegisteredWithTheirGap(t *testing.T) {
 	}
 }
 
-// TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt: object storage is
-// conformant, but a host with no composed object-storage service — or one
-// whose service predates the per-call binding, which work.go leaves unwired —
-// cannot sync a bucket. The catalog says so, and AddSource refuses by name
-// before anything is sealed or stored.
-func TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt(t *testing.T) {
+// TestObjectStorageIsRegisteredOnlyWhereItsServiceResolves: object storage is
+// not compiled into the registry. A host with no composed object-storage
+// service — or one whose service predates the per-call binding, which
+// SetObjectStorage refuses — has no such provider: the catalog does not list
+// it, AddSource refuses a source of it by name before anything is sealed or
+// stored, and an existing source reports why. Resolving the service registers
+// it, conformant and admitted.
+func TestObjectStorageIsRegisteredOnlyWhereItsServiceResolves(t *testing.T) {
 	svc, _ := business.NewService(nil)
 	svc.SetDatasourceConnector(nil, nil, "")
-	accepts := func() bool {
+	listed := func() (present, accepts bool) {
 		for _, e := range svc.DatasourceCatalog() {
 			if e.Descriptor.Key == business.DatasourceProviderUpload {
-				return e.AcceptsNewSources
+				return true, e.AcceptsNewSources
 			}
 		}
-		t.Fatal("object storage is not in the catalog")
-		return false
+		return false, false
 	}
-	if accepts() {
-		t.Fatal("the catalog accepts object-storage sources with no object-storage service composed")
+	if present, _ := listed(); present {
+		t.Fatal("the catalog lists object storage with no object-storage service composed")
 	}
 	_, err := svc.AddSource(context.Background(), "actor-1", business.AddSourceInput{
 		OrgID: "11111111-1111-1111-1111-111111111111", Provider: business.DatasourceProviderUpload, CollectionLabel: "guides", Credential: "secretkey",
@@ -135,8 +138,11 @@ func TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt(t *testing.T) {
 	if status.Code(err) != codes.FailedPrecondition || !strings.Contains(err.Error(), business.ErrObjectStorageUnavailable.Error()) {
 		t.Fatalf("AddSource(upload) with no object storage = %v, want the unavailable refusal", err)
 	}
+	if conformant, gap := business.DatasourceConformance(svc.DatasourceConnectors(), business.DatasourceProviderUpload); conformant || gap != business.ErrObjectStorageUnavailable.Error() {
+		t.Fatalf("an existing object-storage source with no service reports (%v, %q), want the unavailable reason", conformant, gap)
+	}
 	svc.SetDatasourceUploadBucketFactory(func(objectstore.SourceConfig, string, string) objectstore.Bucket { return nil })
-	if !accepts() {
-		t.Fatal("the catalog refuses object-storage sources with an object-storage service composed")
+	if present, accepts := listed(); !present || !accepts {
+		t.Fatalf("with an object-storage service composed the catalog lists object storage = %v, accepting = %v", present, accepts)
 	}
 }

module/services/accounts/code/pkg/business/datasource_object_storage_admission_test.go

@@ -52,10 +52,10 @@ func composedStorageClient(t *testing.T, binding bool) storagev0.ObjectStorageCl
 // review r1's major 3, pinned where the defect was: with the released 0.0.18
 // composed — a service that answers, but cannot serve a binding — the catalog
 // accepted new object-storage sources and AddSource stored them, and only each
-// sync then failed. Wiring that service is refused, and the refusal is at
-// admission: the catalog does not offer the provider, AddSource refuses by
-// name, and nothing is stored or audited. A service that serves the binding is
-// admitted.
+// sync then failed. Wiring that service is refused and registers nothing, so
+// the refusal is at admission: the catalog does not list the provider,
+// AddSource refuses by name, and nothing is stored or audited. A service that
+// serves the binding registers it.
 func TestObjectStorageIsRefusedAtAdmissionWhenTheComposedServiceCannotServeIt(t *testing.T) {
 	ctx := context.Background()
 	accepts := func(svc *business.Service) bool {
@@ -64,7 +64,6 @@ func TestObjectStorageIsRefusedAtAdmissionWhenTheComposedServiceCannotServeIt(t
 				return e.AcceptsNewSources
 			}
 		}
-		t.Fatal("object storage is not in the catalog")
 		return false
 	}
 	input := business.AddSourceInput{