Add support for multiple Pub/Sub notification configurations

fakestorage/server.go

@@ -117,6 +117,11 @@ type Options struct {
 	// of the Google cloud function such events should be published to.
 	EventOptions EventManagerOptions
 
+	// EventConfigs specifies multiple per-bucket Pub/Sub notification
+	// configurations. Each entry defines a topic, a set of event types, and
+	// the bucket it applies to. When non-empty this replaces EventOptions.
+	EventConfigs []EventManagerOptions
+
 	// Location used for buckets in the server.
 	BucketsLocation string
 
@@ -158,10 +163,19 @@ func NewServerWithOptions(options Options) (*Server, error) {
 	s.handler = requestCompressHandler(s.handler)
 	s.transport = &muxTransport{handler: s.handler}
 
-	s.eventManager, err = notification.NewPubsubEventManager(options.EventOptions, options.Writer)
-	if err != nil {
-		return nil, err
+	configs := options.EventConfigs
+	if len(configs) == 0 {
+		configs = []EventManagerOptions{options.EventOptions}
+	}
+	var managers []notification.EventManager
+	for _, cfg := range configs {
+		mgr, err := notification.NewPubsubEventManager(cfg, options.Writer)
+		if err != nil {
+			return nil, err
+		}
+		managers = append(managers, mgr)
 	}
+	s.eventManager = notification.NewMultiEventManager(managers)
 
 	if options.NoListener {
 		return s, nil
@@ -225,7 +239,7 @@ func newServer(options Options) (*Server, error) {
 		externalURL:  options.ExternalURL,
 		publicHost:   publicHost,
 		options:      options,
-		eventManager: &notification.PubsubEventManager{},
+		eventManager: notification.NewMultiEventManager(nil),
 	}
 	s.buildMuxer()
 	_, err = s.seed()

fakestorage/server_test.go

@@ -1076,6 +1076,114 @@ func TestServerEventNotification(t *testing.T) {
 	}
 }
 
+func TestServerMultiEventNotification(t *testing.T) {
+	t.Parallel()
+
+	objA := Object{
+		ObjectAttrs: ObjectAttrs{BucketName: "bucket-a", Name: "file-a.txt"},
+		Content:     []byte("content-a"),
+	}
+	objB := Object{
+		ObjectAttrs: ObjectAttrs{BucketName: "bucket-b", Name: "file-b.txt"},
+		Content:     []byte("content-b"),
+	}
+
+	t.Run("fan-out to multiple managers", func(t *testing.T) {
+		t.Parallel()
+		server, err := NewServerWithOptions(Options{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer server.Stop()
+
+		managerA := &fakeEventManager{}
+		managerB := &fakeEventManager{}
+		server.eventManager = notification.NewMultiEventManager([]notification.EventManager{managerA, managerB})
+
+		if err := server.backend.CreateBucket("bucket-a", backend.BucketAttrs{}); err != nil {
+			t.Fatal(err)
+		}
+		if err := server.backend.CreateBucket("bucket-b", backend.BucketAttrs{}); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := createObjectAction(objA)(server.Client()); err != nil {
+			t.Fatal(err)
+		}
+		if err := createObjectAction(objB)(server.Client()); err != nil {
+			t.Fatal(err)
+		}
+
+		// Both managers should have seen both Finalize events (no bucket filter on fakeEventManager).
+		expectedEvents := []fakeEvent{
+			{obj: fakeEventFieldsFromObject(objA), eventType: notification.EventFinalize},
+			{obj: fakeEventFieldsFromObject(objB), eventType: notification.EventFinalize},
+		}
+		assert.ElementsMatch(t, expectedEvents, managerA.events)
+		assert.ElementsMatch(t, expectedEvents, managerB.events)
+	})
+
+	t.Run("events carry correct bucket names", func(t *testing.T) {
+		t.Parallel()
+		server, err := NewServerWithOptions(Options{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer server.Stop()
+
+		manager := &fakeEventManager{}
+		server.eventManager = notification.NewMultiEventManager([]notification.EventManager{manager})
+
+		if err := server.backend.CreateBucket("bucket-a", backend.BucketAttrs{}); err != nil {
+			t.Fatal(err)
+		}
+		if err := server.backend.CreateBucket("bucket-b", backend.BucketAttrs{}); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := createObjectAction(objA)(server.Client()); err != nil {
+			t.Fatal(err)
+		}
+		if err := createObjectAction(objB)(server.Client()); err != nil {
+			t.Fatal(err)
+		}
+
+		if len(manager.events) != 2 {
+			t.Fatalf("expected 2 events, got %d", len(manager.events))
+		}
+
+		buckets := map[string]bool{}
+		for _, ev := range manager.events {
+			buckets[ev.obj.BucketName] = true
+		}
+		if !buckets["bucket-a"] {
+			t.Error("expected an event for bucket-a")
+		}
+		if !buckets["bucket-b"] {
+			t.Error("expected an event for bucket-b")
+		}
+	})
+
+	t.Run("empty managers list is no-op", func(t *testing.T) {
+		t.Parallel()
+		server, err := NewServerWithOptions(Options{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer server.Stop()
+
+		server.eventManager = notification.NewMultiEventManager(nil)
+
+		if err := server.backend.CreateBucket("bucket-a", backend.BucketAttrs{}); err != nil {
+			t.Fatal(err)
+		}
+		// Must not panic.
+		if err := createObjectAction(objA)(server.Client()); err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
 func TestServerBatchRequest(t *testing.T) {
 	objects := []Object{
 		{

internal/config/config.go

@@ -48,6 +48,8 @@ type Config struct {
 	backend            string
 	fsRoot             string
 	event              EventConfig
+	events             eventConfigFlag
+	parsedEvents       []notification.EventManagerOptions
 	bucketLocation     string
 	LogLevel           slog.Level
 }
@@ -60,6 +62,105 @@ type EventConfig struct {
 	list            []string
 }
 
+// eventConfigFlag is a flag.Value that accumulates repeated -event.config
+// values. Each invocation of Set appends the raw string; validation and
+// conversion into EventManagerOptions happens in Config.validate.
+type eventConfigFlag []string
+
+func (f *eventConfigFlag) String() string {
+	if f == nil {
+		return ""
+	}
+	return strings.Join(*f, ", ")
+}
+
+func (f *eventConfigFlag) Set(value string) error {
+	*f = append(*f, value)
+	return nil
+}
+
+// parseEventConfig parses a single -event.config value into an
+// EventManagerOptions.  The expected format is:
+//
+// bucket=<name>;project=<project>;topic=<topic>[;events=<e1>,<e2>][;prefix=<prefix>]
+//
+// bucket, project, and topic are required. events defaults to finalize if not
+// specified. prefix is optional.
+func parseEventConfig(raw string) (notification.EventManagerOptions, error) {
+	var opts notification.EventManagerOptions
+	var hasBucket, hasProject, hasTopic, hasEvents bool
+
+	for _, field := range strings.Split(raw, ";") {
+		parts := strings.SplitN(field, "=", 2)
+		if len(parts) != 2 {
+			return opts, fmt.Errorf("invalid event config field %q: expected key=value format", field)
+		}
+		key := strings.TrimSpace(parts[0])
+		val := strings.TrimSpace(parts[1])
+
+		switch key {
+		case "bucket":
+			opts.Bucket = val
+			hasBucket = true
+		case "project":
+			opts.ProjectID = val
+			hasProject = true
+		case "topic":
+			opts.TopicName = val
+			hasTopic = true
+		case "events":
+			notifyOn, err := eventListToNotifyOn(strings.Split(val, ","))
+			if err != nil {
+				return opts, err
+			}
+			opts.NotifyOn = notifyOn
+			hasEvents = true
+		case "prefix":
+			opts.ObjectPrefix = val
+		default:
+			return opts, fmt.Errorf("unknown event config key %q", key)
+		}
+	}
+
+	if !hasBucket {
+		return opts, fmt.Errorf("event config missing required field \"bucket\"")
+	}
+	if !hasProject {
+		return opts, fmt.Errorf("event config missing required field \"project\"")
+	}
+	if !hasTopic {
+		return opts, fmt.Errorf("event config missing required field \"topic\"")
+	}
+	if !hasEvents {
+		opts.NotifyOn = notification.EventNotificationOptions{Finalize: true}
+	}
+	return opts, nil
+}
+
+// eventListToNotifyOn validates a list of event name strings and maps them to an
+// EventNotificationOptions struct.
+func eventListToNotifyOn(events []string) (notification.EventNotificationOptions, error) {
+	var notifyOn notification.EventNotificationOptions
+	if len(events) == 0 {
+		return notifyOn, fmt.Errorf("events list must not be empty")
+	}
+	for _, e := range events {
+		switch strings.TrimSpace(e) {
+		case eventFinalize:
+			notifyOn.Finalize = true
+		case eventDelete:
+			notifyOn.Delete = true
+		case eventMetadataUpdate:
+			notifyOn.MetadataUpdate = true
+		case eventArchive:
+			notifyOn.Archive = true
+		default:
+			return notifyOn, fmt.Errorf("%q is an invalid event", strings.TrimSpace(e))
+		}
+	}
+	return notifyOn, nil
+}
+
 // Load parses the given arguments list and return a config object (and/or an
 // error in case of failures).
 func Load(args []string) (Config, error) {
@@ -84,6 +185,7 @@ func Load(args []string) (Config, error) {
 	fs.StringVar(&cfg.event.bucket, "event.bucket", "", "if not empty, only objects in this bucket will generate trigger events")
 	fs.StringVar(&cfg.event.prefix, "event.object-prefix", "", "if not empty, only objects having this prefix will generate trigger events")
 	fs.StringVar(&eventList, "event.list", eventFinalize, "comma separated list of events to publish on cloud function URl. Options are: finalize, delete, and metadataUpdate")
+	fs.Var(&cfg.events, "event.config", "notification configuration in the format: bucket=<bucket>;project=<project>;topic=<topic>[;events=<event1>,<event2>][;prefix=<prefix>]. Can be specified multiple times for multiple configurations. events defaults to finalize. Supported events: finalize, delete, metadataUpdate, archive")
 	fs.StringVar(&cfg.bucketLocation, "location", "US-CENTRAL1", "location for buckets")
 	fs.StringVar(&cfg.CertificateLocation, "cert-location", "", "location for server certificate")
 	fs.StringVar(&cfg.PrivateKeyLocation, "private-key-location", "", "location for private key")
@@ -160,7 +262,19 @@ func (c *Config) validate() error {
 		return fmt.Errorf("port-http %d is too high, maximum value is %d", c.PortHTTP, math.MaxUint16)
 	}
 
-	return c.event.validate()
+	if err := c.event.validate(); err != nil {
+		return err
+	}
+
+	for _, raw := range c.events {
+		opts, err := parseEventConfig(raw)
+		if err != nil {
+			return err
+		}
+		c.parsedEvents = append(c.parsedEvents, opts)
+	}
+
+	return nil
 }
 
 func (c *EventConfig) validate() error {
@@ -230,6 +344,7 @@ func (c *Config) ToFakeGcsOptions(logger *slog.Logger, scheme string) fakestorag
 		AllowedCORSHeaders:  c.allowedCORSHeaders,
 		Writer:              &slogWriter{logger: logger, level: slog.LevelInfo},
 		EventOptions:        eventOptions,
+		EventConfigs:        c.parsedEvents,
 		BucketsLocation:     c.bucketLocation,
 		CertificateLocation: c.CertificateLocation,
 		PrivateKeyLocation:  c.PrivateKeyLocation,