Add runtime per-bucket pub/sub notification configuration API

fakestorage/bucket.go

@@ -166,6 +166,7 @@ func (s *Server) deleteBucket(r *http.Request) jsonResponse {
 	if err != nil {
 		return jsonResponse{status: http.StatusInternalServerError, errorMessage: err.Error()}
 	}
+	s.notificationRegistry.DeleteBucket(bucketName)
 	return jsonResponse{}
 }
 

fakestorage/notification.go

@@ -0,0 +1,65 @@
+package fakestorage
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/fsouza/fake-gcs-server/internal/notification"
+	"github.com/gorilla/mux"
+)
+
+func (s *Server) insertNotification(r *http.Request) jsonResponse {
+	bucketName := unescapeMuxVars(mux.Vars(r))["bucketName"]
+
+	if _, err := s.backend.GetBucket(bucketName); err != nil {
+		return jsonResponse{status: http.StatusNotFound}
+	}
+
+	var cfg notification.NotificationConfig
+	if err := json.NewDecoder(r.Body).Decode(&cfg); err != nil {
+		return jsonResponse{status: http.StatusBadRequest, errorMessage: err.Error()}
+	}
+	if cfg.Topic == "" {
+		return jsonResponse{status: http.StatusBadRequest, errorMessage: "topic is required"}
+	}
+
+	created := s.notificationRegistry.Insert(bucketName, cfg)
+	return jsonResponse{status: http.StatusCreated, data: created}
+}
+
+func (s *Server) getNotification(r *http.Request) jsonResponse {
+	vars := unescapeMuxVars(mux.Vars(r))
+	bucketName := vars["bucketName"]
+	notificationID := vars["notificationId"]
+
+	cfg, ok := s.notificationRegistry.Get(bucketName, notificationID)
+	if !ok {
+		return jsonResponse{status: http.StatusNotFound}
+	}
+	return jsonResponse{data: cfg}
+}
+
+func (s *Server) listNotifications(r *http.Request) jsonResponse {
+	bucketName := unescapeMuxVars(mux.Vars(r))["bucketName"]
+
+	if _, err := s.backend.GetBucket(bucketName); err != nil {
+		return jsonResponse{status: http.StatusNotFound}
+	}
+
+	cfgs := s.notificationRegistry.List(bucketName)
+	if cfgs == nil {
+		cfgs = []notification.NotificationConfig{}
+	}
+	return jsonResponse{data: map[string]interface{}{"kind": "storage#notifications", "items": cfgs}}
+}
+
+func (s *Server) deleteNotification(r *http.Request) jsonResponse {
+	vars := unescapeMuxVars(mux.Vars(r))
+	bucketName := vars["bucketName"]
+	notificationID := vars["notificationId"]
+
+	if !s.notificationRegistry.Delete(bucketName, notificationID) {
+		return jsonResponse{status: http.StatusNotFound}
+	}
+	return jsonResponse{status: http.StatusNoContent}
+}

fakestorage/notification_test.go

@@ -0,0 +1,145 @@
+package fakestorage
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/fsouza/fake-gcs-server/internal/notification"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newNotificationServer(t *testing.T) *Server {
+	t.Helper()
+	srv, err := NewServerWithOptions(Options{NoListener: true})
+	require.NoError(t, err)
+	srv.CreateBucketWithOpts(CreateBucketOpts{Name: "test-bucket"})
+	return srv
+}
+
+func postNotification(t *testing.T, client *http.Client, bucket string, cfg notification.NotificationConfig) (*http.Response, notification.NotificationConfig) {
+	t.Helper()
+	body, _ := json.Marshal(cfg)
+	resp, err := client.Post(
+		fmt.Sprintf("https://storage.googleapis.com/storage/v1/b/%s/notificationConfigs", bucket),
+		"application/json",
+		bytes.NewReader(body),
+	)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	var created notification.NotificationConfig
+	if resp.StatusCode == http.StatusCreated {
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&created))
+	}
+	return resp, created
+}
+
+func TestInsertNotification(t *testing.T) {
+	srv := newNotificationServer(t)
+	cfg := notification.NotificationConfig{Topic: "projects/p/topics/t", PayloadFormat: "JSON_API_V1"}
+
+	resp, created := postNotification(t, srv.HTTPClient(), "test-bucket", cfg)
+	assert.Equal(t, http.StatusCreated, resp.StatusCode)
+	assert.NotEmpty(t, created.ID)
+	assert.Equal(t, "storage#notification", created.Kind)
+	assert.Equal(t, cfg.Topic, created.Topic)
+}
+
+func TestInsertNotification_BucketNotFound(t *testing.T) {
+	srv := newNotificationServer(t)
+	cfg := notification.NotificationConfig{Topic: "projects/p/topics/t"}
+	resp, _ := postNotification(t, srv.HTTPClient(), "no-such-bucket", cfg)
+	assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+}
+
+func TestInsertNotification_MissingTopic(t *testing.T) {
+	srv := newNotificationServer(t)
+	resp, _ := postNotification(t, srv.HTTPClient(), "test-bucket", notification.NotificationConfig{})
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+}
+
+func TestGetNotification(t *testing.T) {
+	srv := newNotificationServer(t)
+	cfg := notification.NotificationConfig{Topic: "projects/p/topics/t"}
+	_, inserted := postNotification(t, srv.HTTPClient(), "test-bucket", cfg)
+
+	resp, err := srv.HTTPClient().Get(fmt.Sprintf(
+		"https://storage.googleapis.com/storage/v1/b/test-bucket/notificationConfigs/%s",
+		inserted.ID,
+	))
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+	var got notification.NotificationConfig
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&got))
+	assert.Equal(t, inserted.ID, got.ID)
+	assert.Equal(t, cfg.Topic, got.Topic)
+}
+
+func TestGetNotification_NotFound(t *testing.T) {
+	srv := newNotificationServer(t)
+	resp, err := srv.HTTPClient().Get(
+		"https://storage.googleapis.com/storage/v1/b/test-bucket/notificationConfigs/9999",
+	)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+}
+
+func TestListNotifications(t *testing.T) {
+	srv := newNotificationServer(t)
+	client := srv.HTTPClient()
+	listURL := "https://storage.googleapis.com/storage/v1/b/test-bucket/notificationConfigs"
+	cfg := notification.NotificationConfig{Topic: "projects/p/topics/t"}
+
+	// empty list
+	resp, err := client.Get(listURL)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	var listResp struct {
+		Items []notification.NotificationConfig `json:"items"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&listResp))
+	assert.Empty(t, listResp.Items)
+
+	// insert two
+	for i := 0; i < 2; i++ {
+		postNotification(t, client, "test-bucket", cfg)
+	}
+
+	resp2, err := client.Get(listURL)
+	require.NoError(t, err)
+	defer resp2.Body.Close()
+	require.NoError(t, json.NewDecoder(resp2.Body).Decode(&listResp))
+	assert.Len(t, listResp.Items, 2)
+}
+
+func TestDeleteNotification(t *testing.T) {
+	srv := newNotificationServer(t)
+	client := srv.HTTPClient()
+	cfg := notification.NotificationConfig{Topic: "projects/p/topics/t"}
+	_, inserted := postNotification(t, client, "test-bucket", cfg)
+
+	deleteURL := fmt.Sprintf(
+		"https://storage.googleapis.com/storage/v1/b/test-bucket/notificationConfigs/%s",
+		inserted.ID,
+	)
+
+	req, _ := http.NewRequest(http.MethodDelete, deleteURL, nil)
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+
+	// second delete → 404
+	req2, _ := http.NewRequest(http.MethodDelete, deleteURL, nil)
+	resp2, err := client.Do(req2)
+	require.NoError(t, err)
+	resp2.Body.Close()
+	assert.Equal(t, http.StatusNotFound, resp2.StatusCode)
+}

fakestorage/object.go

@@ -7,6 +7,7 @@ package fakestorage
 import (
 	"bytes"
 	"compress/gzip"
+	"context"
 	"encoding/json"
 	"encoding/xml"
 	"errors"
@@ -329,13 +330,16 @@ func (s *Server) createObject(obj StreamingObject, conditions backend.Conditions
 		bucket, _ := s.backend.GetBucket(obj.BucketName)
 		if bucket.VersioningEnabled {
 			s.eventManager.Trigger(&oldBackendObj, notification.EventArchive, oldObjEventAttr)
+			s.notificationRegistry.Trigger(context.Background(), &oldBackendObj, notification.EventArchive, oldObjEventAttr)
 		} else {
 			s.eventManager.Trigger(&oldBackendObj, notification.EventDelete, oldObjEventAttr)
+			s.notificationRegistry.Trigger(context.Background(), &oldBackendObj, notification.EventDelete, oldObjEventAttr)
 		}
 	}
 
 	newObj := fromBackendObjects([]backend.StreamingObject{newBackendObj})[0]
 	s.eventManager.Trigger(&newBackendObj, notification.EventFinalize, newObjEventAttr)
+	s.notificationRegistry.Trigger(context.Background(), &newBackendObj, notification.EventFinalize, newObjEventAttr)
 	return newObj, nil
 }
 
@@ -804,8 +808,10 @@ func (s *Server) deleteObject(r *http.Request) jsonResponse {
 	backendObj := toBackendObjects([]StreamingObject{obj})[0]
 	if bucket.VersioningEnabled {
 		s.eventManager.Trigger(&backendObj, notification.EventArchive, nil)
+		s.notificationRegistry.Trigger(context.Background(), &backendObj, notification.EventArchive, nil)
 	} else {
 		s.eventManager.Trigger(&backendObj, notification.EventDelete, nil)
+		s.notificationRegistry.Trigger(context.Background(), &backendObj, notification.EventDelete, nil)
 	}
 	return jsonResponse{}
 }
@@ -1295,6 +1301,7 @@ func (s *Server) patchObject(r *http.Request) jsonResponse {
 	defer backendObj.Close()
 
 	s.eventManager.Trigger(&backendObj, notification.EventMetadata, nil)
+	s.notificationRegistry.Trigger(context.Background(), &backendObj, notification.EventMetadata, nil)
 	return jsonResponse{data: fromBackendObjects([]backend.StreamingObject{backendObj})[0]}
 }
 
@@ -1360,6 +1367,7 @@ func (s *Server) updateObject(r *http.Request) jsonResponse {
 	defer backendObj.Close()
 
 	s.eventManager.Trigger(&backendObj, notification.EventMetadata, nil)
+	s.notificationRegistry.Trigger(context.Background(), &backendObj, notification.EventMetadata, nil)
 	return jsonResponse{data: fromBackendObjects([]backend.StreamingObject{backendObj})[0]}
 }
 
@@ -1417,6 +1425,7 @@ func (s *Server) composeObject(r *http.Request) jsonResponse {
 	obj := fromBackendObjects([]backend.StreamingObject{backendObj})[0]
 
 	s.eventManager.Trigger(&backendObj, notification.EventFinalize, nil)
+	s.notificationRegistry.Trigger(context.Background(), &backendObj, notification.EventFinalize, nil)
 
 	return jsonResponse{data: newObjectResponse(obj.ObjectAttrs, urlhelper.GetBaseURL(r))}
 }

fakestorage/server.go

@@ -42,15 +42,16 @@ const defaultPublicHost = "storage.googleapis.com"
 //
 // It provides a fake implementation of the Google Cloud Storage API.
 type Server struct {
-	backend      backend.Storage
-	uploads      sync.Map
-	transport    *muxTransport
-	ts           *httptest.Server
-	handler      http.Handler
-	options      Options
-	externalURL  string
-	publicHost   string
-	eventManager notification.EventManager
+	backend              backend.Storage
+	uploads              sync.Map
+	transport            *muxTransport
+	ts                   *httptest.Server
+	handler              http.Handler
+	options              Options
+	externalURL          string
+	publicHost           string
+	eventManager         notification.EventManager
+	notificationRegistry *notification.NotificationRegistry
 }
 
 // NewServer creates a new instance of the server, pre-loaded with the given
@@ -220,12 +221,13 @@ func newServer(options Options) (*Server, error) {
 	}
 
 	s := Server{
-		backend:      backendStorage,
-		uploads:      sync.Map{},
-		externalURL:  options.ExternalURL,
-		publicHost:   publicHost,
-		options:      options,
-		eventManager: &notification.PubsubEventManager{},
+		backend:              backendStorage,
+		uploads:              sync.Map{},
+		externalURL:          options.ExternalURL,
+		publicHost:           publicHost,
+		options:              options,
+		eventManager:         &notification.PubsubEventManager{},
+		notificationRegistry: notification.NewNotificationRegistry(options.Writer),
 	}
 	s.buildMuxer()
 	_, err = s.seed()
@@ -282,6 +284,10 @@ func (s *Server) buildMuxer() {
 		r.Path("/b/{sourceBucket}/o/{sourceObject:.+}/{copyType:rewriteTo|copyTo}/b/{destinationBucket}/o/{destinationObject:.+}").Methods(http.MethodPost).HandlerFunc(jsonToHTTPHandler(s.rewriteObject))
 		r.Path("/b/{bucketName}/o/{destinationObject:.+}/compose").Methods(http.MethodPost).HandlerFunc(jsonToHTTPHandler(s.composeObject))
 		r.Path("/b/{bucketName}/o/{objectName:.+}").Methods(http.MethodPut, http.MethodPost).HandlerFunc(jsonToHTTPHandler(s.updateObject))
+		r.Path("/b/{bucketName}/notificationConfigs").Methods(http.MethodPost).HandlerFunc(jsonToHTTPHandler(s.insertNotification))
+		r.Path("/b/{bucketName}/notificationConfigs").Methods(http.MethodGet).HandlerFunc(jsonToHTTPHandler(s.listNotifications))
+		r.Path("/b/{bucketName}/notificationConfigs/{notificationId}").Methods(http.MethodGet).HandlerFunc(jsonToHTTPHandler(s.getNotification))
+		r.Path("/b/{bucketName}/notificationConfigs/{notificationId}").Methods(http.MethodDelete).HandlerFunc(jsonToHTTPHandler(s.deleteNotification))
 	}
 
 	// Internal / update server configuration
@@ -458,6 +464,7 @@ func (s *Server) Stop() {
 	if s.ts != nil {
 		s.ts.Close()
 	}
+	s.notificationRegistry.Close()
 }
 
 // URL returns the server URL.

internal/notification/event.go

@@ -184,6 +184,10 @@ type gcsEvent struct {
 }
 
 func generateEvent(o *backend.StreamingObject, eventType EventType, eventTime string, extraEventAttr map[string]string) ([]byte, map[string]string, error) {
+	return generateEventWithAttrs(o, eventType, eventTime, extraEventAttr, nil)
+}
+
+func generateEventWithAttrs(o *backend.StreamingObject, eventType EventType, eventTime string, extraEventAttr map[string]string, seed map[string]string) ([]byte, map[string]string, error) {
 	payload := gcsEvent{
 		Kind:            "storage#object",
 		ID:              o.ID(),
@@ -200,14 +204,16 @@ func generateEvent(o *backend.StreamingObject, eventType EventType, eventTime st
 		CRC32c:          o.Crc32c,
 		MetaData:        o.Metadata,
 	}
-	attributes := map[string]string{
-		"bucketId":         o.BucketName,
-		"eventTime":        eventTime,
-		"eventType":        string(eventType),
-		"objectGeneration": strconv.FormatInt(o.Generation, 10),
-		"objectId":         o.Name,
-		"payloadFormat":    "JSON_API_V1",
+	attributes := make(map[string]string, len(seed)+6+len(extraEventAttr))
+	for k, v := range seed {
+		attributes[k] = v
 	}
+	attributes["bucketId"] = o.BucketName
+	attributes["eventTime"] = eventTime
+	attributes["eventType"] = string(eventType)
+	attributes["objectGeneration"] = strconv.FormatInt(o.Generation, 10)
+	attributes["objectId"] = o.Name
+	attributes["payloadFormat"] = "JSON_API_V1"
 	for k, v := range extraEventAttr {
 		if _, exists := attributes[k]; exists {
 			return nil, nil, fmt.Errorf("cannot overwrite duplicate event attribute %s", k)