Update CDP code samples using BufferedStorageBackend (#1772)

Author: urvisavla

docs/build/apps/ingest-sdk/ingestion-pipeline-code.mdx

@@ -173,13 +173,13 @@ func (adapter *LedgerMetadataInboundAdapter) Run(ctx context.Context) error {
 
   ledgerRange := ledgerbackend.UnboundedRange(latestNetworkLedger)
 
-  pubConfig := PublisherConfig{
+  pubConfig := ingest.PublisherConfig{
     DataStoreConfig:       adapter.dataStoreConfig,
-    BufferedStorageConfig: DefaultBufferedStorageBackendConfig(adapter.dataStoreConfig.Schema.LedgersPerFile),
+    BufferedStorageConfig: ingest.DefaultBufferedStorageBackendConfig(1),
   }
 
   fmt.Printf("beginning payments stream, starting at ledger %v ...\n", latestNetworkLedger)
-  return ApplyLedgerMetadata(ledgerRange, pubConfig, ctx,
+  return ingest.ApplyLedgerMetadata(ledgerRange, pubConfig, ctx,
       func(lcm xdr.LedgerCloseMeta) error {
           for _, processor := range adapter.processors {
               if err = processor.Process(ctx, Message{Payload: lcm}); err != nil {
@@ -249,10 +249,6 @@ type = "GCS"
 [params]
 destination_bucket_path = "my-gcs-bucketname/prefix1/prefix2"
 
-[schema]
-ledgers_per_file = 1
-files_per_partition = 64000
-
 ```
 
 </CodeExample>

docs/data/indexers/build-your-own/ingest-sdk/developer_guide/ledgerbackends/bufferedstoragebackend.mdx

@@ -109,10 +109,6 @@ func main() {
 		Params: map[string]string{
 			"destination_bucket_path": "your-gcs-bucket/data", // Replace with actual GCS bucket path
 		},
-		Schema: datastore.DataStoreSchema{
-			LedgersPerFile:    1,     // Defines how many ledgers per file
-			FilesPerPartition: 64000, // Defines partition size
-		},
 	}
 
 	// Initialize the datastore
@@ -130,8 +126,15 @@ func main() {
 		RetryWait:  5 * time.Second, // Wait time between retries
 	}
 
+	// Per SEP-54 (https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0054.md),
+	// the schema is stored in the datastore; LoadSchema retrieves it.
+	schema, err := datastore.LoadSchema(context.Background(), dataStore, datastoreConfig)
+	if err != nil {
+		log.Fatal(errors.Wrap(err, "failed to retrieve datastore schema"))
+	}
+
 	// Initialize the backend
-	backend, err := ledgerbackend.NewBufferedStorageBackend(backendConfig, dataStore)
+	backend, err := ledgerbackend.NewBufferedStorageBackend(backendConfig, dataStore, schema)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed to create buffered storage backend"))
 	}