Repository: google/exposure-notifications-server
Pull request #397 (2020-05-21 04:37:12)

Filesystem support
Making Blobstore configurable via environment variables.
The feature allows configuring different blob store systems for different installations.

google-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/397#" title="Author self-approved">mgulimonov</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mikehelmick**
You can assign the PR to them by writing `/assign @mikehelmick` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/google/exposure-notifications-server/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["mikehelmick"]} -->
mgulimonov: /assign @mikehelmick

Patch:
diff --git a/cmd/monolith/main.go b/cmd/monolith/main.go
index a1e0581..9461689 100644
--- a/cmd/monolith/main.go
+++ b/cmd/monolith/main.go
@@ -29,6 +29,7 @@ import (
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/publish"
 	"github.com/google/exposure-notifications-server/internal/setup"
+	"github.com/google/exposure-notifications-server/internal/storage"
 )
 
 var _ setup.DBConfigProvider = (*MonoConfig)(nil)
@@ -45,11 +46,16 @@ type MonoConfig struct {
 	Publish       *publish.Config
 	Database      *database.Config
 	FederationIn  *federationin.Config
+	BlobstoreType string `envconfig:"BLOBSTORE_TYPE" default:"CLOUD_STORAGE"`
 }
 
-func (c *MonoConfig) DB() *database.Config                       { return c.Database }
-func (c *MonoConfig) KeyManager() bool                           { return true }
-func (c *MonoConfig) BlobStorage() bool                          { return true }
+func (c *MonoConfig) DB() *database.Config { return c.Database }
+func (c *MonoConfig) KeyManager() bool     { return true }
+func (c *MonoConfig) BlobStorage() storage.BlobstoreConfig {
+	return storage.BlobstoreConfig{
+		BlobstoreType: storage.BlobstoreType(c.BlobstoreType),
+	}
+}
 func (c *MonoConfig) AuthorizedAppConfig() *authorizedapp.Config { return c.AuthorizedApp }
 
 func main() {
diff --git a/internal/authorizedapp/database/authorized_app.go b/internal/authorizedapp/database/authorized_app.go
index d7f5ec3..8bc3c68 100644
--- a/internal/authorizedapp/database/authorized_app.go
+++ b/internal/authorizedapp/database/authorized_app.go
@@ -31,7 +31,7 @@ type AuthorizedAppDB struct {
 	db *database.DB
 }
 
-func NewAuthorizedAppDB(db *database.DB) *AuthorizedAppDB {
+func New(db *database.DB) *AuthorizedAppDB {
 	return &AuthorizedAppDB{
 		db: db,
 	}
diff --git a/internal/authorizedapp/database_provider.go b/internal/authorizedapp/database_provider.go
index b67b48e..a47c00a 100644
--- a/internal/authorizedapp/database_provider.go
+++ b/internal/authorizedapp/database_provider.go
@@ -143,7 +143,7 @@ func (p *DatabaseProvider) loadAuthorizedAppFromDatabase(ctx context.Context, na
 	logger := logging.FromContext(ctx)
 
 	logger.Infof("authorizedapp: loading %v from database", name)
-	config, err := authorizedappdb.NewAuthorizedAppDB(p.database).GetAuthorizedApp(ctx, p.secretManager, name)
+	config, err := authorizedappdb.New(p.database).GetAuthorizedApp(ctx, p.secretManager, name)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read %v from database: %w", name, err)
 	}
diff --git a/internal/cleanup/cleanup.go b/internal/cleanup/cleanup.go
index 66bb729..d242497 100644
--- a/internal/cleanup/cleanup.go
+++ b/internal/cleanup/cleanup.go
@@ -21,7 +21,9 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/database"
+	publishdb "github.com/google/exposure-notifications-server/internal/publish/database"
+
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 	"github.com/google/exposure-notifications-server/internal/storage"
@@ -41,14 +43,14 @@ func NewExposureHandler(config *Config, env *serverenv.ServerEnv) (http.Handler,
 	return &exposureCleanupHandler{
 		config:   config,
 		env:      env,
-		database: env.Database(),
+		database: publishdb.New(env.Database()),
 	}, nil
 }
 
 type exposureCleanupHandler struct {
 	config   *Config
 	env      *serverenv.ServerEnv
-	database *database.DB
+	database *publishdb.PublishDB
 }
 
 func (h *exposureCleanupHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -96,7 +98,7 @@ func NewExportHandler(config *Config, env *serverenv.ServerEnv) (http.Handler, e
 	return &exportCleanupHandler{
 		config:    config,
 		env:       env,
-		database:  env.Database(),
+		database:  database.New(env.Database()),
 		blobstore: env.Blobstore(),
 	}, nil
 }
@@ -104,7 +106,7 @@ func NewExportHandler(config *Config, env *serverenv.ServerEnv) (http.Handler, e
 type exportCleanupHandler struct {
 	config    *Config
 	env       *serverenv.ServerEnv
-	database  *database.DB
+	database  *database.ExportDB
 	blobstore storage.Blobstore
 }
 
diff --git a/internal/cleanup/config.go b/internal/cleanup/config.go
index f600154..9de7908 100644
--- a/internal/cleanup/config.go
+++ b/internal/cleanup/config.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/setup"
+	"github.com/google/exposure-notifications-server/internal/storage"
 )
 
 // Compile-time check to assert this config matches requirements.
@@ -28,10 +29,11 @@ var _ setup.DBConfigProvider = (*Config)(nil)
 // Config represents the configuration and associated environment variables for
 // the cleanup components.
 type Config struct {
-	Port     string        `envconfig:"PORT" default:"8080"`
-	Timeout  time.Duration `envconfig:"CLEANUP_TIMEOUT" default:"10m"`
-	TTL      time.Duration `envconfig:"CLEANUP_TTL" default:"336h"`
-	Database *database.Config
+	Port          string        `envconfig:"PORT" default:"8080"`
+	Timeout       time.Duration `envconfig:"CLEANUP_TIMEOUT" default:"10m"`
+	TTL           time.Duration `envconfig:"CLEANUP_TTL" default:"336h"`
+	Database      *database.Config
+	BlobstoreType string `envconfig:"BLOBSTORE_TYPE" default:"CLOUD_STORAGE"`
 }
 
 // DB return the databsae configuration.
@@ -40,6 +42,8 @@ func (c *Config) DB() *database.Config {
 }
 
 // BlobStorage returns the BlobStorage configuration.
-func (c *Config) BlobStorage() bool {
-	return true
+func (c *Config) BlobStorage() storage.BlobstoreConfig {
+	return storage.BlobstoreConfig{
+		BlobstoreType: storage.BlobstoreType(c.BlobstoreType),
+	}
 }
diff --git a/internal/database/database.go b/internal/database/database.go
index 7572db4..87f5fd2 100644
--- a/internal/database/database.go
+++ b/internal/database/database.go
@@ -41,8 +41,8 @@ func toNullString(s string) sql.NullString {
 	}
 }
 
-// inTx runs the given function f within a transaction with isolation level isoLevel.
-func (db *DB) inTx(ctx context.Context, isoLevel pgx.TxIsoLevel, f func(tx pgx.Tx) error) error {
+// InTx runs the given function f within a transaction with isolation level isoLevel.
+func (db *DB) InTx(ctx context.Context, isoLevel pgx.TxIsoLevel, f func(tx pgx.Tx) error) error {
 	conn, err := db.Pool.Acquire(ctx)
 	if err != nil {
 		return fmt.Errorf("acquiring connection: %v", err)
diff --git a/internal/database/federationin.go b/internal/database/federationin.go
index 21b7870..22ee0d8 100644
--- a/internal/database/federationin.go
+++ b/internal/database/federationin.go
@@ -61,7 +61,7 @@ func getFederationInQuery(ctx context.Context, queryID string, queryRow queryRow
 
 // AddFederationInQuery adds a FederationInQuery entity. It will overwrite a query with matching q.queryID if it exists.
 func (db *DB) AddFederationInQuery(ctx context.Context, q *FederationInQuery) error {
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	return db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		query := `
 			INSERT INTO
 				FederationInQuery
@@ -153,7 +153,7 @@ func (db *DB) StartFederationInSync(ctx context.Context, q *FederationInQuery, s
 	finalize := func(maxTimestamp time.Time, totalInserted int) error {
 		completed := started.Add(time.Now().Sub(startedTimer))
 
-		return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		return db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 			// Special case: when no keys are pulled, the maxTimestamp will be 0, so we don't update the
 			// FederationQuery in this case to prevent it from going back and fetching old keys from the past.
 			if totalInserted > 0 {
diff --git a/internal/database/federationout.go b/internal/database/federationout.go
index b7cff7b..6887a54 100644
--- a/internal/database/federationout.go
+++ b/internal/database/federationout.go
@@ -23,7 +23,7 @@ import (
 
 // AddFederationOutAuthorization adds or updates a FederationOutAuthorization record.
 func (db *DB) AddFederationOutAuthorization(ctx context.Context, auth *FederationOutAuthorization) error {
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	return db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		q := `
 			INSERT INTO
 				FederationOutAuthorization
diff --git a/internal/database/lock.go b/internal/database/lock.go
index e1c42fc..b37dd49 100644
--- a/internal/database/lock.go
+++ b/internal/database/lock.go
@@ -38,7 +38,7 @@ type UnlockFn func() error
 // Lock acquires lock with given name that times out after ttl. Returns an UnlockFn that can be used to unlock the lock. ErrAlreadyLocked will be returned if there is already a lock in use.
 func (db *DB) Lock(ctx context.Context, lockID string, ttl time.Duration) (UnlockFn, error) {
 	var expires time.Time
-	err := db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	err := db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		row := tx.QueryRow(ctx, `
 			SELECT AcquireLock($1, $2)
 		`, lockID, int(ttl.Seconds()))
@@ -59,7 +59,7 @@ func (db *DB) Lock(ctx context.Context, lockID string, ttl time.Duration) (Unloc
 
 func makeUnlockFn(ctx context.Context, db *DB, lockID string, expires time.Time) UnlockFn {
 	return func() error {
-		return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		return db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 			row := tx.QueryRow(ctx, `
 				SELECT ReleaseLock($1, $2)
 			`, lockID, expires)
diff --git a/internal/export/batcher.go b/internal/export/batcher.go
index 279e942..00d77e6 100644
--- a/internal/export/batcher.go
+++ b/internal/export/batcher.go
@@ -22,7 +22,11 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/database"
+	"github.com/google/exposure-notifications-server/internal/export/model"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/logging"
 )
 
@@ -38,7 +42,7 @@ func (s *Server) CreateBatchesHandler(w http.ResponseWriter, r *http.Request) {
 	lock := "create_batches"
 	unlockFn, err := s.db.Lock(ctx, lock, s.config.CreateTimeout)
 	if err != nil {
-		if errors.Is(err, database.ErrAlreadyLocked) {
+		if errors.Is(err, coredb.ErrAlreadyLocked) {
 			metrics.WriteInt("export-batcher-lock-contention", true, 1)
 			msg := fmt.Sprintf("Lock %s already in use, no work will be performed", lock)
 			logger.Infof(msg)
@@ -59,7 +63,7 @@ func (s *Server) CreateBatchesHandler(w http.ResponseWriter, r *http.Request) {
 	}()
 
 	effectiveTime := time.Now().Add(-1 * s.config.MinWindowAge)
-	err = s.db.IterateExportConfigs(ctx, effectiveTime, func(ec *database.ExportConfig) error {
+	err = database.New(s.db).IterateExportConfigs(ctx, effectiveTime, func(ec *model.ExportConfig) error {
 		totalConfigs++
 		if batchesCreated, err := s.maybeCreateBatches(ctx, ec, effectiveTime); err != nil {
 			logger.Errorf("Failed to create batches for config %d: %v, continuing to next config", ec.ConfigID, err)
@@ -88,11 +92,11 @@ func (s *Server) CreateBatchesHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (s *Server) maybeCreateBatches(ctx context.Context, ec *database.ExportConfig, now time.Time) (int, error) {
+func (s *Server) maybeCreateBatches(ctx context.Context, ec *model.ExportConfig, now time.Time) (int, error) {
 	logger := logging.FromContext(ctx)
 	metrics := s.env.MetricsExporter(ctx)
 
-	latestEnd, err := s.db.LatestExportBatchEnd(ctx, ec)
+	latestEnd, err := database.New(s.db).LatestExportBatchEnd(ctx, ec)
 	if err != nil {
 		return 0, fmt.Errorf("fetching most recent batch for config %d: %w", ec.ConfigID, err)
 	}
@@ -104,23 +108,23 @@ func (s *Server) maybeCreateBatches(ctx context.Context, ec *database.ExportConf
 		return 0, nil
 	}
 
-	var batches []*database.ExportBatch
+	var batches []*model.ExportBatch
 	for _, br := range ranges {
 		infoIds := make([]int64, len(ec.SignatureInfoIDs))
 		copy(infoIds, ec.SignatureInfoIDs)
-		batches = append(batches, &database.ExportBatch{
+		batches = append(batches, &model.ExportBatch{
 			ConfigID:         ec.ConfigID,
 			BucketName:       ec.BucketName,
 			FilenameRoot:     ec.FilenameRoot,
 			StartTimestamp:   br.start,
 			EndTimestamp:     br.end,
 			Region:           ec.Region,
-			Status:           database.ExportBatchOpen,
+			Status:           model.ExportBatchOpen,
 			SignatureInfoIDs: infoIds,
 		})
 	}
 
-	if err := s.db.AddExportBatches(ctx, batches); err != nil {
+	if err := database.New(s.db).AddExportBatches(ctx, batches); err != nil {
 		return 0, fmt.Errorf("creating export batches for config %d: %w", ec.ConfigID, err)
 	}
 
@@ -138,7 +142,7 @@ var sanityDate = time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC)
 func makeBatchRanges(period time.Duration, latestEnd, now time.Time, truncateWindow time.Duration) []batchRange {
 
 	// Compute the end of the exposure publish window; we don't want any batches with an end date greater than this time.
-	publishEnd := database.TruncateWindow(now, truncateWindow)
+	publishEnd := publishmodel.TruncateWindow(now, truncateWindow)
 
 	// Special case: if there have not been batches before, return only a single one.
 	// We use sanityDate here because the loop below will happily create batch ranges
diff --git a/internal/export/config.go b/internal/export/config.go
index bc5a8be..aa5a98e 100644
--- a/internal/export/config.go
+++ b/internal/export/config.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/setup"
+	"github.com/google/exposure-notifications-server/internal/storage"
 )
 
 // Compile-time check to assert this config matches requirements.
@@ -38,6 +39,7 @@ type Config struct {
 	MaxRecords     int           `envconfig:"EXPORT_FILE_MAX_RECORDS" default:"30000"`
 	TruncateWindow time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
 	MinWindowAge   time.Duration `envconfig:"MIN_WINDOW_AGE" default:"2h"`
+	BlobstoreType  string        `envconfig:"BLOBSTORE_TYPE" default:"CLOUD_STORAGE"`
 }
 
 // DB returns the database config.
@@ -51,6 +53,8 @@ func (c *Config) KeyManager() bool {
 }
 
 // BlobStorage returns the BlobStorage configuration.
-func (c *Config) BlobStorage() bool {
-	return true
+func (c *Config) BlobStorage() storage.BlobstoreConfig {
+	return storage.BlobstoreConfig{
+		BlobstoreType: storage.BlobstoreType(c.BlobstoreType),
+	}
 }
diff --git a/internal/database/export.go b/internal/export/database/export.go
similarity index 80%
rename from internal/database/export.go
rename to internal/export/database/export.go
index 3ed8137..ba783a8 100644
--- a/internal/database/export.go
+++ b/internal/export/database/export.go
@@ -21,6 +21,8 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/model"
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/storage"
 
@@ -32,8 +34,18 @@ const (
 	oneDay       = 24 * time.Hour
 )
 
+type ExportDB struct {
+	db *database.DB
+}
+
+func New(db *database.DB) *ExportDB {
+	return &ExportDB{
+		db: db,
+	}
+}
+
 // AddExportConfig creates a new ExportConfig record from which batch jobs are created.
-func (db *DB) AddExportConfig(ctx context.Context, ec *ExportConfig) error {
+func (db *ExportDB) AddExportConfig(ctx context.Context, ec *model.ExportConfig) error {
 	if ec.Period > oneDay {
 		return errors.New("maximum period is 24h")
 	}
@@ -48,7 +60,7 @@ func (db *DB) AddExportConfig(ctx context.Context, ec *ExportConfig) error {
 	if !ec.Thru.IsZero() {
 		thru = &ec.Thru
 	}
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	return db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		row := tx.QueryRow(ctx, `
 			INSERT INTO
 				ExportConfig
@@ -69,14 +81,14 @@ func (db *DB) AddExportConfig(ctx context.Context, ec *ExportConfig) error {
 // IterateExportConfigs applies f to each ExportConfig whose FromTimestamp is
 // before the given time. If f returns a non-nil error, the iteration stops, and
 // the returned error will match f's error with errors.Is.
-func (db *DB) IterateExportConfigs(ctx context.Context, t time.Time, f func(*ExportConfig) error) (err error) {
+func (db *ExportDB) IterateExportConfigs(ctx context.Context, t time.Time, f func(*model.ExportConfig) error) (err error) {
 	defer func() {
 		if err != nil {
 			err = fmt.Errorf("IterateExportConfigs(%s): %w", t, err)
 		}
 	}()
 
-	conn, err := db.Pool.Acquire(ctx)
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -98,7 +110,7 @@ func (db *DB) IterateExportConfigs(ctx context.Context, t time.Time, f func(*Exp
 	defer rows.Close()
 	for rows.Next() {
 		var (
-			m             ExportConfig
+			m             model.ExportConfig
 			periodSeconds int
 			thru          *time.Time
 		)
@@ -116,7 +128,7 @@ func (db *DB) IterateExportConfigs(ctx context.Context, t time.Time, f func(*Exp
 	return rows.Err()
 }
 
-func (db *DB) AddSignatureInfo(ctx context.Context, si *SignatureInfo) error {
+func (db *ExportDB) AddSignatureInfo(ctx context.Context, si *model.SignatureInfo) error {
 	if si.SigningKey == "" {
 		return fmt.Errorf("signing key cannot be empty for a signature info")
 	}
@@ -125,7 +137,7 @@ func (db *DB) AddSignatureInfo(ctx context.Context, si *SignatureInfo) error {
 	if !si.EndTimestamp.IsZero() {
 		thru = &si.EndTimestamp
 	}
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	return db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		row := tx.QueryRow(ctx, `
       INSERT INTO
         SignatureInfo
@@ -142,8 +154,8 @@ func (db *DB) AddSignatureInfo(ctx context.Context, si *SignatureInfo) error {
 	})
 }
 
-func (db *DB) LookupSignatureInfos(ctx context.Context, ids []int64, validUntil time.Time) ([]*SignatureInfo, error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *ExportDB) LookupSignatureInfos(ctx context.Context, ids []int64, validUntil time.Time) ([]*model.SignatureInfo, error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -161,12 +173,12 @@ func (db *DB) LookupSignatureInfos(ctx context.Context, ids []int64, validUntil
 		return nil, err
 	}
 
-	var sigInfos []*SignatureInfo
+	var sigInfos []*model.SignatureInfo
 	for rows.Next() {
 		if rows.Err() != nil {
 			return nil, rows.Err()
 		}
-		var info SignatureInfo
+		var info model.SignatureInfo
 		var thru *time.Time
 		if err := rows.Scan(&info.ID, &info.SigningKey, &info.AppPackageName, &info.BundleID, &info.SigningKeyVersion, &info.SigningKeyID, &thru); err != nil {
 			return nil, err
@@ -184,8 +196,8 @@ func (db *DB) LookupSignatureInfos(ctx context.Context, ids []int64, validUntil
 // a given ExportConfig. It returns the zero time if no previous ExportBatch
 // exists.
 // TODO(squee1945): This needs a
-func (db *DB) LatestExportBatchEnd(ctx context.Context, ec *ExportConfig) (time.Time, error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *ExportDB) LatestExportBatchEnd(ctx context.Context, ec *model.ExportConfig) (time.Time, error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return time.Time{}, fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -214,8 +226,8 @@ func (db *DB) LatestExportBatchEnd(ctx context.Context, ec *ExportConfig) (time.
 }
 
 // AddExportBatches inserts new export batches.
-func (db *DB) AddExportBatches(ctx context.Context, batches []*ExportBatch) error {
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+func (db *ExportDB) AddExportBatches(ctx context.Context, batches []*model.ExportBatch) error {
+	return db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		const stmtName = "insert export batches"
 		_, err := tx.Prepare(ctx, stmtName, `
 			INSERT INTO
@@ -239,11 +251,11 @@ func (db *DB) AddExportBatches(ctx context.Context, batches []*ExportBatch) erro
 }
 
 // LeaseBatch returns a leased ExportBatch for the worker to process. If no work to do, nil will be returned.
-func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time) (*ExportBatch, error) {
+func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time) (*model.ExportBatch, error) {
 	// Lookup a set of candidate batch IDs.
 	var openBatchIDs []int64
 	err := func() error { // Use a func to allow defer conn.Release() to work.
-		conn, err := db.Pool.Acquire(ctx)
+		conn, err := db.db.Pool.Acquire(ctx)
 		if err != nil {
 			return fmt.Errorf("acquiring connection: %w", err)
 		}
@@ -265,7 +277,7 @@ func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time)
 			AND
 				end_timestamp < $3
 			LIMIT 100
-		`, ExportBatchOpen, ExportBatchPending, now)
+		`, model.ExportBatchOpen, model.ExportBatchPending, now)
 		if err != nil {
 			return err
 		}
@@ -297,7 +309,7 @@ func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time)
 	for _, bid := range openBatchIDs {
 		// In a serialized transaction, fetch the existing batch and make sure it can be leased, then lease it.
 		leased := false
-		err := db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		err := db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 
 			row := tx.QueryRow(ctx, `
 				SELECT
@@ -314,7 +326,7 @@ func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time)
 				return err
 			}
 
-			if status == ExportBatchComplete || (expires != nil && status == ExportBatchPending && now.Before(*expires)) {
+			if status == model.ExportBatchComplete || (expires != nil && status == model.ExportBatchPending && now.Before(*expires)) {
 				// Something beat us to this batch, it's no longer available.
 				return nil
 			}
@@ -326,7 +338,7 @@ func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time)
 					status = $1, lease_expires = $2
 				WHERE
 				    batch_id = $3
-				`, ExportBatchPending, now.Add(ttl), bid)
+				`, model.ExportBatchPending, now.Add(ttl), bid)
 			if err != nil {
 				return err
 			}
@@ -347,8 +359,8 @@ func (db *DB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time)
 }
 
 // LookupExportBatch returns an ExportBatch for the given batchID.
-func (db *DB) LookupExportBatch(ctx context.Context, batchID int64) (*ExportBatch, error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *ExportDB) LookupExportBatch(ctx context.Context, batchID int64) (*model.ExportBatch, error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -357,7 +369,9 @@ func (db *DB) LookupExportBatch(ctx context.Context, batchID int64) (*ExportBatc
 	return lookupExportBatch(ctx, batchID, conn.QueryRow)
 }
 
-func lookupExportBatch(ctx context.Context, batchID int64, queryRow queryRowFn) (*ExportBatch, error) {
+type queryRowFn func(ctx context.Context, query string, args ...interface{}) pgx.Row
+
+func lookupExportBatch(ctx context.Context, batchID int64, queryRow queryRowFn) (*model.ExportBatch, error) {
 	row := queryRow(ctx, `
 		SELECT
 			batch_id, config_id, bucket_name, filename_root, start_timestamp, end_timestamp, region, status, lease_expires, signature_info_ids
@@ -369,10 +383,10 @@ func lookupExportBatch(ctx context.Context, batchID int64, queryRow queryRowFn)
 		`, batchID)
 
 	var expires *time.Time
-	eb := ExportBatch{}
+	eb := model.ExportBatch{}
 	if err := row.Scan(&eb.BatchID, &eb.ConfigID, &eb.BucketName, &eb.FilenameRoot, &eb.StartTimestamp, &eb.EndTimestamp, &eb.Region, &eb.Status, &expires, &eb.SignatureInfoIDs); err != nil {
 		if err == pgx.ErrNoRows {
-			return nil, ErrNotFound
+			return nil, database.ErrNotFound
 		}
 		return nil, err
 	}
@@ -383,21 +397,21 @@ func lookupExportBatch(ctx context.Context, batchID int64, queryRow queryRowFn)
 }
 
 // FinalizeBatch writes the ExportFile records and marks the ExportBatch as complete.
-func (db *DB) FinalizeBatch(ctx context.Context, eb *ExportBatch, files []string, batchSize int) error {
-	return db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+func (db *ExportDB) FinalizeBatch(ctx context.Context, eb *model.ExportBatch, files []string, batchSize int) error {
+	return db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		// Update ExportFile for the files created.
 		for i, file := range files {
-			ef := ExportFile{
+			ef := model.ExportFile{
 				BucketName: eb.BucketName,
 				Filename:   file,
 				BatchID:    eb.BatchID,
 				Region:     eb.Region,
 				BatchNum:   i + 1,
 				BatchSize:  batchSize,
-				Status:     ExportBatchComplete,
+				Status:     model.ExportBatchComplete,
 			}
 			if err := addExportFile(ctx, tx, &ef); err != nil {
-				if err == ErrKeyConflict {
+				if err == database.ErrKeyConflict {
 					logging.FromContext(ctx).Infof("ExportFile %q already exists in database, skipping without overwriting. This can occur when reprocessing a failed batch.", file)
 				} else {
 					return fmt.Errorf("adding export file entry: %w", err)
@@ -414,8 +428,8 @@ func (db *DB) FinalizeBatch(ctx context.Context, eb *ExportBatch, files []string
 }
 
 // LookupExportFiles returns a list of export files for the given ExportConfig exportConfigID.
-func (db *DB) LookupExportFiles(ctx context.Context, exportConfigID int64) ([]string, error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *ExportDB) LookupExportFiles(ctx context.Context, exportConfigID int64) ([]string, error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -434,7 +448,7 @@ func (db *DB) LookupExportFiles(ctx context.Context, exportConfigID int64) ([]st
 			eb.status = $2
 		ORDER BY
 			ef.filename
-		`, exportConfigID, ExportBatchComplete)
+		`, exportConfigID, model.ExportBatchComplete)
 	if err != nil {
 		return nil, err
 	}
@@ -463,8 +477,8 @@ type joinedExportBatchFile struct {
 	batchStatus string
 }
 
-func (db *DB) LookupExportFile(ctx context.Context, filename string) (*ExportFile, error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *ExportDB) LookupExportFile(ctx context.Context, filename string) (*model.ExportFile, error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("acquiring connection: %w", err)
 	}
@@ -480,10 +494,10 @@ func (db *DB) LookupExportFile(ctx context.Context, filename string) (*ExportFil
 		LIMIT 1
 		`, filename)
 
-	ef := ExportFile{}
+	ef := model.ExportFile{}
 	if err := row.Scan(&ef.BucketName, &ef.Filename, &ef.BatchID, &ef.Region, &ef.BatchNum, &ef.BatchSize, &ef.Status); err != nil {
 		if err == pgx.ErrNoRows {
-			return nil, ErrNotFound
+			return nil, database.ErrNotFound
 		}
 		return nil, err
 	}
@@ -491,13 +505,13 @@ func (db *DB) LookupExportFile(ctx context.Context, filename string) (*ExportFil
 }
 
 // DeleteFilesBefore deletes the export batch files for batches ending before the time passed in.
-func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore storage.Blobstore) (int, error) {
+func (db *ExportDB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore storage.Blobstore) (int, error) {
 	logger := logging.FromContext(ctx)
 
 	// Fetch filenames for  batches where at least one file is not deleted yet.
 	var files []joinedExportBatchFile
 	err := func() error { // Use a func to allow defer conn.Release() to work.
-		conn, err := db.Pool.Acquire(ctx)
+		conn, err := db.db.Pool.Acquire(ctx)
 		if err != nil {
 			return fmt.Errorf("acquiring connection: %w", err)
 		}
@@ -518,7 +532,7 @@ func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore
 			WHERE
 				eb.end_timestamp < $1
 				AND eb.status != $2`
-		rows, err := conn.Query(ctx, q, before, ExportBatchDeleted)
+		rows, err := conn.Query(ctx, q, before, model.ExportBatchDeleted)
 		if err != nil {
 			return fmt.Errorf("fetching filenames: %w", err)
 		}
@@ -543,7 +557,7 @@ func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore
 	for _, f := range files {
 
 		// If file is already deleted, skip to the next.
-		if f.fileStatus == ExportBatchDeleted {
+		if f.fileStatus == model.ExportBatchDeleted {
 			batchFileDeleteCounter[f.batchID]++
 			continue
 		}
@@ -555,15 +569,15 @@ func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore
 			return 0, fmt.Errorf("delete object: %w", err)
 		}
 
-		err := db.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		err := db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 			// Update Status in ExportFile.
-			if err := updateExportFileStatus(ctx, tx, f.batchID, f.filename, ExportBatchDeleted); err != nil {
+			if err := updateExportFileStatus(ctx, tx, f.batchID, f.filename, model.ExportBatchDeleted); err != nil {
 				return fmt.Errorf("updating ExportFile: %w", err)
 			}
 
 			// If batch completely deleted, update in ExportBatch.
 			if batchFileDeleteCounter[f.batchID] == f.count {
-				if err := updateExportBatchStatus(ctx, tx, f.batchID, ExportBatchDeleted); err != nil {
+				if err := updateExportBatchStatus(ctx, tx, f.batchID, model.ExportBatchDeleted); err != nil {
 					return fmt.Errorf("updating ExportBatch: %w", err)
 				}
 			}
@@ -582,7 +596,7 @@ func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore
 
 // addExportFile adds a row to ExportFile. If the row already exists (based on the primary key),
 // ErrKeyConflict is returned.
-func addExportFile(ctx context.Context, tx pgx.Tx, ef *ExportFile) error {
+func addExportFile(ctx context.Context, tx pgx.Tx, ef *model.ExportFile) error {
 	tag, err := tx.Exec(ctx, `
 		INSERT INTO
 			ExportFile
@@ -595,7 +609,7 @@ func addExportFile(ctx context.Context, tx pgx.Tx, ef *ExportFile) error {
 		return fmt.Errorf("inserting to ExportFile: %w", err)
 	}
 	if tag.RowsAffected() == 0 {
-		return ErrKeyConflict
+		return database.ErrKeyConflict
 	}
 	return nil
 }
@@ -638,9 +652,9 @@ func completeBatch(ctx context.Context, tx pgx.Tx, batchID int64) error {
 		return err
 	}
 
-	if batch.Status == ExportBatchComplete {
+	if batch.Status == model.ExportBatchComplete {
 		// Batch is already completed.
-		logger.Warnf("When completing a batch, the status of batch %d was already %s.", batchID, ExportBatchComplete)
+		logger.Warnf("When completing a batch, the status of batch %d was already %s.", batchID, model.ExportBatchComplete)
 		return nil
 	}
 
@@ -651,7 +665,7 @@ func completeBatch(ctx context.Context, tx pgx.Tx, batchID int64) error {
 			status = $1, lease_expires = NULL
 		WHERE
 			batch_id = $2
-		`, ExportBatchComplete, batchID)
+		`, model.ExportBatchComplete, batchID)
 	if err != nil {
 		return err
 	}
diff --git a/internal/export/exportfile.go b/internal/export/exportfile.go
index fe8bd62..c3c7fc3 100644
--- a/internal/export/exportfile.go
+++ b/internal/export/exportfile.go
@@ -23,7 +23,9 @@ import (
 	"fmt"
 	"sort"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/model"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/pb/export"
 
 	"github.com/golang/protobuf/proto"
@@ -39,12 +41,12 @@ const (
 )
 
 type ExportSigners struct {
-	SignatureInfo *database.SignatureInfo
+	SignatureInfo *model.SignatureInfo
 	Signer        crypto.Signer
 }
 
 // MarshalExportFile converts the inputs into an encoded byte array.
-func MarshalExportFile(eb *database.ExportBatch, exposures []*database.Exposure, batchNum, batchSize int, signers []ExportSigners) ([]byte, error) {
+func MarshalExportFile(eb *model.ExportBatch, exposures []*publishmodel.Exposure, batchNum, batchSize int, signers []ExportSigners) ([]byte, error) {
 	// create main exposure key export binary
 	expContents, err := marshalContents(eb, exposures, int32(batchNum), int32(batchSize), signers)
 	if err != nil {
@@ -82,7 +84,7 @@ func MarshalExportFile(eb *database.ExportBatch, exposures []*database.Exposure,
 	return buf.Bytes(), nil
 }
 
-func marshalContents(eb *database.ExportBatch, exposures []*database.Exposure, batchNum int32, batchSize int32, signers []ExportSigners) ([]byte, error) {
+func marshalContents(eb *model.ExportBatch, exposures []*publishmodel.Exposure, batchNum int32, batchSize int32, signers []ExportSigners) ([]byte, error) {
 	exportBytes := []byte("EK Export v1    ")
 	if len(exportBytes) != fixedHeaderWidth {
 		return nil, fmt.Errorf("incorrect header length: %d", len(exportBytes))
@@ -129,7 +131,7 @@ func marshalContents(eb *database.ExportBatch, exposures []*database.Exposure, b
 	return append(exportBytes, protoBytes...), nil
 }
 
-func createSignatureInfo(si *database.SignatureInfo) *export.SignatureInfo {
+func createSignatureInfo(si *model.SignatureInfo) *export.SignatureInfo {
 	sigInfo := &export.SignatureInfo{SignatureAlgorithm: proto.String(algorithm)}
 	if si.AppPackageName != "" {
 		sigInfo.AndroidPackage = proto.String(si.AppPackageName)
@@ -146,7 +148,7 @@ func createSignatureInfo(si *database.SignatureInfo) *export.SignatureInfo {
 	return sigInfo
 }
 
-func marshalSignature(eb *database.ExportBatch, exportContents []byte, batchNum, batchSize int32, signers []ExportSigners) ([]byte, error) {
+func marshalSignature(eb *model.ExportBatch, exportContents []byte, batchNum, batchSize int32, signers []ExportSigners) ([]byte, error) {
 
 	var signatures []*export.TEKSignature
 	for _, s := range signers {
diff --git a/internal/database/export_model.go b/internal/export/model/export_model.go
similarity index 99%
rename from internal/database/export_model.go
rename to internal/export/model/export_model.go
index 33c066f..79729da 100644
--- a/internal/database/export_model.go
+++ b/internal/export/model/export_model.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package database
+package model
 
 import (
 	"time"
diff --git a/internal/export/server.go b/internal/export/server.go
index c914d12..e759787 100644
--- a/internal/export/server.go
+++ b/internal/export/server.go
@@ -17,7 +17,10 @@ package export
 import (
 	"fmt"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/database"
+	publishdb "github.com/google/exposure-notifications-server/internal/publish/database"
+
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 )
 
@@ -35,15 +38,19 @@ func NewServer(config *Config, env *serverenv.ServerEnv) (*Server, error) {
 	}
 
 	return &Server{
-		db:     env.Database(),
-		config: config,
-		env:    env,
+		db:        env.Database(),
+		exportdb:  database.New(env.Database()),
+		publishdb: publishdb.New(env.Database()),
+		config:    config,
+		env:       env,
 	}, nil
 }
 
 // Server hosts end points to manage export batches.
 type Server struct {
-	db     *database.DB
-	config *Config
-	env    *serverenv.ServerEnv
+	db        *coredb.DB
+	exportdb  *database.ExportDB
+	publishdb *publishdb.PublishDB
+	config    *Config
+	env       *serverenv.ServerEnv
 }
diff --git a/internal/export/worker.go b/internal/export/worker.go
index abdb1f0..e10751a 100644
--- a/internal/export/worker.go
+++ b/internal/export/worker.go
@@ -24,7 +24,13 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/database"
+	publishdb "github.com/google/exposure-notifications-server/internal/publish/database"
+
+	"github.com/google/exposure-notifications-server/internal/export/model"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/util"
 )
@@ -39,6 +45,7 @@ func (s *Server) WorkerHandler(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), s.config.WorkerTimeout)
 	defer cancel()
 	logger := logging.FromContext(ctx)
+	exportDB := database.New(s.db)
 
 	emitIndexForEmptyBatch := true
 	for {
@@ -53,7 +60,7 @@ func (s *Server) WorkerHandler(w http.ResponseWriter, r *http.Request) {
 		minutesAgo := time.Now().Add(-5 * time.Minute)
 
 		// Check for a batch and obtain a lease for it.
-		batch, err := s.db.LeaseBatch(ctx, s.config.WorkerTimeout, minutesAgo)
+		batch, err := exportDB.LeaseBatch(ctx, s.config.WorkerTimeout, minutesAgo)
 		if err != nil {
 			logger.Errorf("Failed to lease batch: %v", err)
 			continue
@@ -79,11 +86,11 @@ func (s *Server) WorkerHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emitIndexForEmptyBatch bool) error {
+func (s *Server) exportBatch(ctx context.Context, eb *model.ExportBatch, emitIndexForEmptyBatch bool) error {
 	logger := logging.FromContext(ctx)
 	logger.Infof("Processing export batch %d (root: %q, region: %s), max records per file %d", eb.BatchID, eb.FilenameRoot, eb.Region, s.config.MaxRecords)
 
-	criteria := database.IterateExposuresCriteria{
+	criteria := publishdb.IterateExposuresCriteria{
 		SinceTimestamp:      eb.StartTimestamp,
 		UntilTimestamp:      eb.EndTimestamp,
 		IncludeRegions:      []string{eb.Region},
@@ -93,9 +100,10 @@ func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emit
 	// Build up groups of exposures in memory. We need to use memory so we can determine the
 	// total number of groups (which is embedded in each export file). This technique avoids
 	// SELECT COUNT which would lock the database slowing new uploads.
-	var groups [][]*database.Exposure
-	var exposures []*database.Exposure
-	_, err := s.db.IterateExposures(ctx, criteria, func(exp *database.Exposure) error {
+	var groups [][]*publishmodel.Exposure
+	var exposures []*publishmodel.Exposure
+
+	_, err := s.publishdb.IterateExposures(ctx, criteria, func(exp *publishmodel.Exposure) error {
 		exposures = append(exposures, exp)
 		if len(exposures) == s.config.MaxRecords {
 			groups = append(groups, exposures)
@@ -103,6 +111,7 @@ func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emit
 		}
 		return nil
 	})
+
 	if err != nil {
 		return fmt.Errorf("iterating exposures: %w", err)
 	}
@@ -121,7 +130,7 @@ func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emit
 	}
 
 	// Load the non-expired signature infos associated with this export batch.
-	sigInfos, err := s.db.LookupSignatureInfos(ctx, eb.SignatureInfoIDs, time.Now())
+	sigInfos, err := s.exportdb.LookupSignatureInfos(ctx, eb.SignatureInfoIDs, time.Now())
 	if err != nil {
 		return fmt.Errorf("error loading signature info for batch %d, %w", eb.BatchID, err)
 	}
@@ -159,7 +168,7 @@ func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emit
 	}
 
 	// Write the files records in database and complete the batch.
-	if err := s.db.FinalizeBatch(ctx, eb, objectNames, batchSize); err != nil {
+	if err := s.exportdb.FinalizeBatch(ctx, eb, objectNames, batchSize); err != nil {
 		return fmt.Errorf("completing batch: %w", err)
 	}
 	logger.Infof("Batch %d completed", eb.BatchID)
@@ -167,9 +176,9 @@ func (s *Server) exportBatch(ctx context.Context, eb *database.ExportBatch, emit
 }
 
 type createFileInfo struct {
-	exposures      []*database.Exposure
-	exportBatch    *database.ExportBatch
-	signatureInfos []*database.SignatureInfo
+	exposures      []*publishmodel.Exposure
+	exportBatch    *model.ExportBatch
+	signatureInfos []*model.SignatureInfo
 	batchNum       int
 	batchSize      int
 }
@@ -205,7 +214,7 @@ func (s *Server) createFile(ctx context.Context, cfi createFileInfo) (string, er
 
 // retryingCreateIndex create the index file. The index file includes _all_ batches for an ExportConfig,
 // so multiple workers may be racing to update it. We use a lock to make them line up after one another.
-func (s *Server) retryingCreateIndex(ctx context.Context, eb *database.ExportBatch, objectNames []string) error {
+func (s *Server) retryingCreateIndex(ctx context.Context, eb *model.ExportBatch, objectNames []string) error {
 	logger := logging.FromContext(ctx)
 
 	lockID := fmt.Sprintf("export-batch-%d", eb.BatchID)
@@ -218,7 +227,7 @@ func (s *Server) retryingCreateIndex(ctx context.Context, eb *database.ExportBat
 
 		unlock, err := s.db.Lock(ctx, lockID, time.Minute)
 		if err != nil {
-			if err == database.ErrAlreadyLocked {
+			if err == coredb.ErrAlreadyLocked {
 				logger.Debugf("Lock %s is locked; sleeping %v and will try again", lockID, sleep)
 				time.Sleep(sleep)
 				continue
@@ -242,8 +251,9 @@ func (s *Server) retryingCreateIndex(ctx context.Context, eb *database.ExportBat
 	return nil
 }
 
-func (s *Server) createIndex(ctx context.Context, eb *database.ExportBatch, newObjectNames []string) (string, int, error) {
-	objects, err := s.db.LookupExportFiles(ctx, eb.ConfigID)
+func (s *Server) createIndex(ctx context.Context, eb *model.ExportBatch, newObjectNames []string) (string, int, error) {
+	exportDB := database.New(s.db)
+	objects, err := exportDB.LookupExportFiles(ctx, eb.ConfigID)
 	if err != nil {
 		return "", 0, fmt.Errorf("lookup existing export files for batch %d: %w", eb.BatchID, err)
 	}
@@ -273,11 +283,11 @@ func (s *Server) createIndex(ctx context.Context, eb *database.ExportBatch, newO
 	return indexObjectName, len(objects), nil
 }
 
-func exportFilename(eb *database.ExportBatch, batchNum int) string {
+func exportFilename(eb *model.ExportBatch, batchNum int) string {
 	return fmt.Sprintf("%s/%d-%05d%s", eb.FilenameRoot, eb.StartTimestamp.Unix(), batchNum, filenameSuffix)
 }
 
-func exportIndexFilename(eb *database.ExportBatch) string {
+func exportIndexFilename(eb *model.ExportBatch) string {
 	return fmt.Sprintf("%s/index.txt", eb.FilenameRoot)
 }
 
@@ -290,7 +300,7 @@ func randomInt(min, max int) (int, error) {
 	return int(n.Int64()) + min, nil
 }
 
-func ensureMinNumExposures(exposures []*database.Exposure, region string, minLength, jitter int) ([]*database.Exposure, error) {
+func ensureMinNumExposures(exposures []*publishmodel.Exposure, region string, minLength, jitter int) ([]*publishmodel.Exposure, error) {
 	if len(exposures) == 0 {
 		return exposures, nil
 	}
@@ -302,7 +312,7 @@ func ensureMinNumExposures(exposures []*database.Exposure, region string, minLen
 		// Pieces needed are
 		// (1) exposure key, (2) interval number, (3) transmission risk
 		// Exposure key is 16 random bytes.
-		eKey := make([]byte, database.KeyLength)
+		eKey := make([]byte, publishmodel.KeyLength)
 		_, err := rand.Read(eKey)
 		if err != nil {
 			return nil, fmt.Errorf("rand.Read: %w", err)
@@ -328,13 +338,13 @@ func ensureMinNumExposures(exposures []*database.Exposure, region string, minLen
 		}
 		intervalCount := exposures[fromIdx].IntervalCount
 
-		ek := &database.Exposure{
+		ek := &publishmodel.Exposure{
 			ExposureKey:      eKey,
 			TransmissionRisk: transmissionRisk,
 			Regions:          []string{region},
 			IntervalNumber:   intervalNumber,
 			IntervalCount:    intervalCount,
-			// The rest of the database.Exposure fields are not used in the export file.
+			// The rest of the publishmodel.Exposure fields are not used in the export file.
 		}
 		exposures = append(exposures, ek)
 	}
diff --git a/internal/federationin/federationin.go b/internal/federationin/federationin.go
index 0faee37..1eb3163 100644
--- a/internal/federationin/federationin.go
+++ b/internal/federationin/federationin.go
@@ -27,10 +27,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	coredb "github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/metrics"
 	"github.com/google/exposure-notifications-server/internal/pb"
+	"github.com/google/exposure-notifications-server/internal/publish/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 
 	"google.golang.org/api/idtoken"
@@ -49,8 +51,8 @@ var (
 
 type (
 	fetchFn               func(context.Context, *pb.FederationFetchRequest, ...grpc.CallOption) (*pb.FederationFetchResponse, error)
-	insertExposuresFn     func(context.Context, []*database.Exposure) error
-	startFederationSyncFn func(context.Context, *database.FederationInQuery, time.Time) (int64, database.FinalizeSyncFn, error)
+	insertExposuresFn     func(context.Context, []*model.Exposure) error
+	startFederationSyncFn func(context.Context, *coredb.FederationInQuery, time.Time) (int64, coredb.FinalizeSyncFn, error)
 )
 
 type pullDependencies struct {
@@ -63,16 +65,18 @@ type pullDependencies struct {
 // federation results for a single federation query.
 func NewHandler(env *serverenv.ServerEnv, config *Config) http.Handler {
 	return &handler{
-		env:    env,
-		db:     env.Database(),
-		config: config,
+		env:       env,
+		db:        env.Database(),
+		publishdb: database.New(env.Database()),
+		config:    config,
 	}
 }
 
 type handler struct {
-	env    *serverenv.ServerEnv
-	db     *database.DB
-	config *Config
+	env       *serverenv.ServerEnv
+	db        *coredb.DB
+	publishdb *database.PublishDB
+	config    *Config
 }
 
 func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -102,7 +106,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	lock := "query_" + queryID
 	unlockFn, err := h.db.Lock(ctx, lock, h.config.Timeout)
 	if err != nil {
-		if errors.Is(err, database.ErrAlreadyLocked) {
+		if errors.Is(err, coredb.ErrAlreadyLocked) {
 			metrics.WriteInt("federation-pull-lock-contention", true, 1)
 			msg := fmt.Sprintf("Lock %s already in use. No work will be performed.", lock)
 			logger.Infof(msg)
@@ -116,7 +120,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	query, err := h.db.GetFederationInQuery(ctx, queryID)
 	if err != nil {
-		if errors.Is(err, database.ErrNotFound) {
+		if errors.Is(err, coredb.ErrNotFound) {
 			badRequestf(ctx, w, "unknown %s", queryParam)
 			return
 		}
@@ -170,7 +174,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	deps := pullDependencies{
 		fetch:               client.Fetch,
-		insertExposures:     h.db.InsertExposures,
+		insertExposures:     h.publishdb.InsertExposures,
 		startFederationSync: h.db.StartFederationInSync,
 	}
 	batchStart := time.Now()
@@ -184,7 +188,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies, q *database.FederationInQuery, batchStart time.Time, truncateWindow time.Duration) error {
+func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies, q *coredb.FederationInQuery, batchStart time.Time, truncateWindow time.Duration) error {
 	logger := logging.FromContext(ctx)
 	logger.Infof("Processing query %q", q.QueryID)
 
@@ -205,7 +209,7 @@ func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies,
 		logger.Infof("Inserted %d keys", total)
 	}()
 
-	createdAt := database.TruncateWindow(batchStart, truncateWindow)
+	createdAt := model.TruncateWindow(batchStart, truncateWindow)
 	partial := true
 	for partial {
 
@@ -222,7 +226,7 @@ func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies,
 		}
 
 		// Loop through the result set, storing in database.
-		var exposures []*database.Exposure
+		var exposures []*model.Exposure
 		for _, ctr := range response.Response {
 
 			var upperRegions []string
@@ -234,12 +238,12 @@ func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies,
 			for _, cti := range ctr.ContactTracingInfo {
 				for _, key := range cti.ExposureKeys {
 
-					if cti.TransmissionRisk < database.MinTransmissionRisk || cti.TransmissionRisk > database.MaxTransmissionRisk {
+					if cti.TransmissionRisk < model.MinTransmissionRisk || cti.TransmissionRisk > model.MaxTransmissionRisk {
 						logger.Errorf("invalid transmission risk %v - dropping record.", cti.TransmissionRisk)
 						continue
 					}
 
-					exposures = append(exposures, &database.Exposure{
+					exposures = append(exposures, &model.Exposure{
 						TransmissionRisk: int(cti.TransmissionRisk),
 						ExposureKey:      key.ExposureKey,
 						Regions:          upperRegions,
diff --git a/internal/federationout/federationout.go b/internal/federationout/federationout.go
index 2170e74..b4c8817 100644
--- a/internal/federationout/federationout.go
+++ b/internal/federationout/federationout.go
@@ -23,9 +23,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/logging"
 	"github.com/google/exposure-notifications-server/internal/pb"
+
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 	"google.golang.org/api/idtoken"
 	"google.golang.org/grpc"
@@ -42,21 +45,23 @@ const (
 // Compile time assert that this server implements the required grpc interface.
 var _ pb.FederationServer = (*Server)(nil)
 
-type iterateExposuresFunc func(context.Context, database.IterateExposuresCriteria, func(*database.Exposure) error) (string, error)
+type iterateExposuresFunc func(context.Context, database.IterateExposuresCriteria, func(*model.Exposure) error) (string, error)
 
 // NewServer builds a new FederationServer.
 func NewServer(env *serverenv.ServerEnv, config *Config) pb.FederationServer {
 	return &Server{
-		env:    env,
-		db:     env.Database(),
-		config: config,
+		env:       env,
+		db:        env.Database(),
+		publishdb: database.New(env.Database()),
+		config:    config,
 	}
 }
 
 type Server struct {
-	env    *serverenv.ServerEnv
-	db     *database.DB
-	config *Config
+	env       *serverenv.ServerEnv
+	db        *coredb.DB
+	publishdb *database.PublishDB
+	config    *Config
 }
 
 type authKey struct{}
@@ -66,7 +71,7 @@ func (s Server) Fetch(ctx context.Context, req *pb.FederationFetchRequest) (*pb.
 	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
 	defer cancel()
 	logger := logging.FromContext(ctx)
-	response, err := s.fetch(ctx, req, s.db.IterateExposures, database.TruncateWindow(time.Now(), s.config.TruncateWindow)) // Don't fetch the current window, which isn't complete yet. TODO(squee1945): should I double this for safety?
+	response, err := s.fetch(ctx, req, s.publishdb.IterateExposures, model.TruncateWindow(time.Now(), s.config.TruncateWindow)) // Don't fetch the current window, which isn't complete yet. TODO(squee1945): should I double this for safety?
 	if err != nil {
 		s.env.MetricsExporter(ctx).WriteInt("federation-fetch-failed", true, 1)
 		logger.Errorf("Fetch error: %v", err)
@@ -91,7 +96,7 @@ func (s Server) fetch(ctx context.Context, req *pb.FederationFetchRequest, itFun
 	logger.Infof("Processing client request %#v", req)
 
 	// If there is a FederationAuthorization on the context, set the query to operate within its limits.
-	if auth, ok := ctx.Value(authKey{}).(*database.FederationOutAuthorization); ok {
+	if auth, ok := ctx.Value(authKey{}).(*coredb.FederationOutAuthorization); ok {
 		// For included regions, we INTERSECT the requested included regions with the configured included regions.
 		req.RegionIdentifiers = intersect(req.RegionIdentifiers, auth.IncludeRegions)
 		// For excluded regions, we UNION the the requested excluded regions with the configured excluded regions.
@@ -125,7 +130,7 @@ func (s Server) fetch(ctx context.Context, req *pb.FederationFetchRequest, itFun
 	ctiMap := map[string]*pb.ContactTracingInfo{}     // local index into the response being assembled; keys on unique set of (ctrMap key, transmissionRisk, verificationAuthorityName)
 	response := &pb.FederationFetchResponse{}
 	count := 0
-	cursor, err := itFunc(ctx, criteria, func(inf *database.Exposure) error {
+	cursor, err := itFunc(ctx, criteria, func(inf *model.Exposure) error {
 		// If the diagnosis key is empty, it's malformed, so skip it.
 		if len(inf.ExposureKey) == 0 {
 			logger.Debugf("Exposure %s missing ExposureKey, skipping.", inf.ExposureKey)
@@ -243,7 +248,7 @@ func (s Server) AuthInterceptor(ctx context.Context, req interface{}, info *grpc
 
 	auth, err := s.db.GetFederationOutAuthorization(ctx, token.Issuer, token.Subject)
 	if err != nil {
-		if errors.Is(err, database.ErrNotFound) {
+		if errors.Is(err, coredb.ErrNotFound) {
 			metrics.WriteInt("federation-fetch-unauthorized", true, 1)
 			logger.Infof("Authorization not found (issuer %q, subject %s)", token.Issuer, token.Subject)
 			return nil, status.Errorf(codes.Unauthenticated, "Invalid issuer/subject")
diff --git a/internal/ios/devicecheck.go b/internal/ios/devicecheck.go
index 0961090..4594e6d 100644
--- a/internal/ios/devicecheck.go
+++ b/internal/ios/devicecheck.go
@@ -36,8 +36,13 @@ var (
 	// endpoint is the production Apple DeviceCheck endpoint.
 	endpoint = "https://api.devicecheck.apple.com/v1/validate_device_token"
 
-	// httpTimeout is the maximum amount of time to wait for a response.
-	httpTimeout = 5 * time.Second
+	// The Client's Transport typically has internal state (cached TCP connections), so
+	// Clients should be reused instead of created as needed. Clients are safe for concurrent
+	// use by multiple goroutines.
+	httpClient = &http.Client{
+		// the maximum amount of time to wait for a response
+		Timeout: 5 * time.Second,
+	}
 )
 
 type VerifyOpts struct {
@@ -85,8 +90,7 @@ func ValidateDeviceToken(ctx context.Context, deviceToken string, opts *VerifyOp
 	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", signedJwt))
 
 	// Call Apple's API.
-	client := &http.Client{Timeout: httpTimeout}
-	resp, err := client.Do(req)
+	resp, err := httpClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("failed to make request: %w", err)
 	}
diff --git a/internal/jsonutil/unmarshal.go b/internal/jsonutil/unmarshal.go
index bcc122e..999ba6b 100644
--- a/internal/jsonutil/unmarshal.go
+++ b/internal/jsonutil/unmarshal.go
@@ -33,7 +33,7 @@ const (
 
 // Unmarshal provides a common implementation of JSON unmarshalling with well defined error handling.
 func Unmarshal(w http.ResponseWriter, r *http.Request, data interface{}) (int, error) {
-	if t := r.Header.Get("Content-type"); t != "application/json" {
+	if t := r.Header.Get("content-type"); len(t) < 16 || t[:16] != "application/json" {
 		return http.StatusUnsupportedMediaType, fmt.Errorf("content-type is not application/json")
 	}
 
diff --git a/internal/pb/federation.pb.go b/internal/pb/federation.pb.go
index 11eec2b..3923714 100644
--- a/internal/pb/federation.pb.go
+++ b/internal/pb/federation.pb.go
@@ -255,6 +255,7 @@ type ContactTracingInfo struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
+	// Transmission risk is an integer with valid values from 1-8.
 	TransmissionRisk int32          `protobuf:"varint,1,opt,name=transmissionRisk,proto3" json:"transmissionRisk,omitempty"` // required
 	ExposureKeys     []*ExposureKey `protobuf:"bytes,2,rep,name=exposureKeys,proto3" json:"exposureKeys,omitempty"`
 }
diff --git a/internal/database/exposure.go b/internal/publish/database/exposure.go
similarity index 90%
rename from internal/database/exposure.go
rename to internal/publish/database/exposure.go
index aba2247..8c07898 100644
--- a/internal/database/exposure.go
+++ b/internal/publish/database/exposure.go
@@ -23,7 +23,9 @@ import (
 	"time"
 
 	"github.com/google/exposure-notifications-server/internal/base64util"
+	"github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/logging"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 
 	pgx "github.com/jackc/pgx/v4"
 )
@@ -33,6 +35,16 @@ const (
 	InsertExposuresBatchSize = 500
 )
 
+type PublishDB struct {
+	db *database.DB
+}
+
+func New(db *database.DB) *PublishDB {
+	return &PublishDB{
+		db: db,
+	}
+}
+
 // IterateExposuresCriteria is criteria to iterate exposures.
 type IterateExposuresCriteria struct {
 	IncludeRegions []string
@@ -54,8 +66,8 @@ type IterateExposuresCriteria struct {
 // criteria.LastCursor in a subsequent call to IterateExposures, will continue
 // the iteration at the failed row. If IterateExposures returns a nil error,
 // the first return value will be the empty string.
-func (db *DB) IterateExposures(ctx context.Context, criteria IterateExposuresCriteria, f func(*Exposure) error) (cur string, err error) {
-	conn, err := db.Pool.Acquire(ctx)
+func (db *PublishDB) IterateExposures(ctx context.Context, criteria IterateExposuresCriteria, f func(*model.Exposure) error) (cur string, err error) {
+	conn, err := db.db.Pool.Acquire(ctx)
 	if err != nil {
 		return "", fmt.Errorf("acquiring connection: %v", err)
 	}
@@ -94,7 +106,7 @@ func (db *DB) IterateExposures(ctx context.Context, criteria IterateExposuresCri
 			return cursor(), err
 		}
 		var (
-			m          Exposure
+			m          model.Exposure
 			encodedKey string
 			syncID     *int64
 		)
@@ -178,8 +190,8 @@ func generateExposureQuery(criteria IterateExposuresCriteria) (string, []interfa
 }
 
 // InsertExposures inserts a set of exposures.
-func (db *DB) InsertExposures(ctx context.Context, exposures []*Exposure) error {
-	return db.inTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+func (db *PublishDB) InsertExposures(ctx context.Context, exposures []*model.Exposure) error {
+	return db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
 		const stmtName = "insert exposures"
 		_, err := tx.Prepare(ctx, stmtName, `
 			INSERT INTO
@@ -210,10 +222,10 @@ func (db *DB) InsertExposures(ctx context.Context, exposures []*Exposure) error
 }
 
 // DeleteExposures deletes exposures created before "before" date. Returns the number of records deleted.
-func (db *DB) DeleteExposures(ctx context.Context, before time.Time) (int64, error) {
+func (db *PublishDB) DeleteExposures(ctx context.Context, before time.Time) (int64, error) {
 	var count int64
 	// ReadCommitted is sufficient here because we are dealing with historical, immutable rows.
-	err := db.inTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+	err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
 		result, err := tx.Exec(ctx, `
 			DELETE FROM
 				Exposure
diff --git a/internal/database/exposure_model.go b/internal/publish/model/exposure_model.go
similarity index 99%
rename from internal/database/exposure_model.go
rename to internal/publish/model/exposure_model.go
index e5dae2b..611aab0 100644
--- a/internal/database/exposure_model.go
+++ b/internal/publish/model/exposure_model.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package database
+package model
 
 import (
 	"crypto/sha256"
diff --git a/internal/publish/publish.go b/internal/publish/publish.go
index b8240b1..8338fd0 100644
--- a/internal/publish/publish.go
+++ b/internal/publish/publish.go
@@ -22,9 +22,11 @@ import (
 	"time"
 
 	"github.com/google/exposure-notifications-server/internal/authorizedapp"
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/database"
+
 	"github.com/google/exposure-notifications-server/internal/jsonutil"
 	"github.com/google/exposure-notifications-server/internal/logging"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 	"github.com/google/exposure-notifications-server/internal/verification"
 )
@@ -40,9 +42,9 @@ func NewHandler(ctx context.Context, config *Config, env *serverenv.ServerEnv) (
 		return nil, fmt.Errorf("missing AuthorizedApp provider in server environment")
 	}
 
-	transformer, err := database.NewTransformer(config.MaxKeysOnPublish, config.MaxIntervalAge, config.TruncateWindow)
+	transformer, err := model.NewTransformer(config.MaxKeysOnPublish, config.MaxIntervalAge, config.TruncateWindow)
 	if err != nil {
-		return nil, fmt.Errorf("database.NewTransformer: %w", err)
+		return nil, fmt.Errorf("model.NewTransformer: %w", err)
 	}
 	logger.Infof("max keys per upload: %v", config.MaxKeysOnPublish)
 	logger.Infof("max interval start age: %v", config.MaxIntervalAge)
@@ -52,7 +54,7 @@ func NewHandler(ctx context.Context, config *Config, env *serverenv.ServerEnv) (
 		serverenv:             env,
 		transformer:           transformer,
 		config:                config,
-		database:              env.Database(),
+		database:              database.New(env.Database()),
 		authorizedAppProvider: env.AuthorizedAppProvider(),
 	}, nil
 }
@@ -60,8 +62,8 @@ func NewHandler(ctx context.Context, config *Config, env *serverenv.ServerEnv) (
 type publishHandler struct {
 	config                *Config
 	serverenv             *serverenv.ServerEnv
-	transformer           *database.Transformer
-	database              *database.DB
+	transformer           *model.Transformer
+	database              *database.PublishDB
 	authorizedAppProvider authorizedapp.Provider
 }
 
@@ -77,7 +79,7 @@ func (h *publishHandler) handleRequest(w http.ResponseWriter, r *http.Request) r
 	ctx := r.Context()
 	logger := logging.FromContext(ctx)
 
-	var data *database.Publish
+	var data model.Publish
 	code, err := jsonutil.Unmarshal(w, r, &data)
 	if err != nil {
 		// Log the unparsable JSON, but return success to the client.
@@ -110,7 +112,7 @@ func (h *publishHandler) handleRequest(w http.ResponseWriter, r *http.Request) r
 		}
 	}
 
-	if err := verification.VerifyRegions(appConfig, data); err != nil {
+	if err := verification.VerifyRegions(appConfig, &data); err != nil {
 		message := fmt.Sprintf("verifying allowed regions: %v", err)
 		return response{status: http.StatusUnauthorized, message: message, metric: "publish-region-not-authorized", count: 1}
 	}
@@ -119,7 +121,7 @@ func (h *publishHandler) handleRequest(w http.ResponseWriter, r *http.Request) r
 		if appConfig.DeviceCheckDisabled {
 			logger.Errorf("skipping DeviceCheck for %v (disabled)", data.AppPackageName)
 			h.serverenv.MetricsExporter(ctx).WriteInt("publish-devicecheck-skip", true, 1)
-		} else if err := verification.VerifyDeviceCheck(ctx, appConfig, data); err != nil {
+		} else if err := verification.VerifyDeviceCheck(ctx, appConfig, &data); err != nil {
 			message := fmt.Sprintf("unable to verify devicecheck payload: %v", err)
 			logger.Error(message)
 			return response{status: http.StatusUnauthorized, message: message, metric: "publish-devicecheck-invalid", count: 1}
@@ -128,7 +130,7 @@ func (h *publishHandler) handleRequest(w http.ResponseWriter, r *http.Request) r
 		if appConfig.SafetyNetDisabled {
 			logger.Errorf("skipping SafetyNet for %v (disabled)", data.AppPackageName)
 			h.serverenv.MetricsExporter(ctx).WriteInt("publish-safetynet-skip", true, 1)
-		} else if err := verification.VerifySafetyNet(ctx, time.Now(), appConfig, data); err != nil {
+		} else if err := verification.VerifySafetyNet(ctx, time.Now(), appConfig, &data); err != nil {
 			message := fmt.Sprintf("unable to verify safetynet payload: %v", err)
 			logger.Error(message)
 			return response{status: http.StatusUnauthorized, message: message, metric: "publish-safetnet-invalid", count: 1}
@@ -140,7 +142,7 @@ func (h *publishHandler) handleRequest(w http.ResponseWriter, r *http.Request) r
 	}
 
 	batchTime := time.Now()
-	exposures, err := h.transformer.TransformPublish(data, batchTime)
+	exposures, err := h.transformer.TransformPublish(&data, batchTime)
 	if err != nil {
 		message := fmt.Sprintf("unable to read request data: %v", err)
 		logger.Error(message)
diff --git a/internal/setup/setup.go b/internal/setup/setup.go
index 80772b4..82b1040 100644
--- a/internal/setup/setup.go
+++ b/internal/setup/setup.go
@@ -47,9 +47,9 @@ type KeyManagerProvider interface {
 	KeyManager() bool
 }
 
-// BlobStorageConfigProvider is a marker interface indicating the BlobStorage interface should be installed.
+// BlobStorageConfigProvider provides the information about current Blobstore configuration.
 type BlobStorageConfigProvider interface {
-	BlobStorage() bool
+	BlobStorage() storage.BlobstoreConfig
 }
 
 // Function returned from setup to be deferred until the caller exits.
@@ -86,13 +86,15 @@ func Setup(ctx context.Context, config DBConfigProvider) (*serverenv.ServerEnv,
 		}
 		opts = append(opts, serverenv.WithKeyManager(km))
 	}
-	// TODO(mikehelmick): Make this extensible to other providers.
+
 	if _, ok := config.(BlobStorageConfigProvider); ok {
-		storage, err := storage.NewGoogleCloudStorage(ctx)
+		provider := config.(BlobStorageConfigProvider)
+		blobStore, err := storage.CreateBlobstore(ctx, provider.BlobStorage())
 		if err != nil {
 			return nil, nil, fmt.Errorf("unable to connect to storage system: %v", err)
 		}
-		opts = append(opts, serverenv.WithBlobStorage(storage))
+		blobStorage := serverenv.WithBlobStorage(blobStore)
+		opts = append(opts, blobStorage)
 	}
 
 	// Setup the database connection.
diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index f1b9d1c..80ea5c5 100644
--- a/internal/storage/storage.go
+++ b/internal/storage/storage.go
@@ -15,7 +15,26 @@
 // Package storage is an interface over file/blob storage
 package storage
 
-import "context"
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/exposure-notifications-server/internal/logging"
+)
+
+// Identifies the type of Blobestore to use
+type BlobstoreType string
+
+const (
+	None          BlobstoreType = "NONE"
+	Cloud_storage BlobstoreType = "CLOUD_STORAGE"
+	Filesystem    BlobstoreType = "FILESYSTEM"
+)
+
+// Blobstore Configuration
+type BlobstoreConfig struct {
+	BlobstoreType BlobstoreType
+}
 
 // Blobstore defines the minimum interface for a blob storage system.
 type Blobstore interface {
@@ -25,3 +44,37 @@ type Blobstore interface {
 	// DeleteObject deltes an object or does nothing if the object doesn't exist.
 	DeleteObject(ctx context.Context, bucket, objectName string) error
 }
+
+// Blobstore that does nothing.
+type NoopBlobstore struct{}
+
+func NewNoopBlobstore(ctx context.Context) (Blobstore, error) {
+	return &NoopBlobstore{}, nil
+}
+
+// No op.
+func (s *NoopBlobstore) CreateObject(ctx context.Context, folder, filename string, contents []byte) error {
+	return nil
+}
+
+// No op.
+func (s *NoopBlobstore) DeleteObject(ctx context.Context, folder, filename string) error {
+	return nil
+}
+
+// Creates a new BlobstoreFactory based on the provided BlobstoreType.
+func CreateBlobstore(ctx context.Context, config BlobstoreConfig) (Blobstore, error) {
+	logger := logging.FromContext(ctx)
+	logger.Infof("BlobstoreType is set up to %v", config.BlobstoreType)
+
+	switch config.BlobstoreType {
+	case Cloud_storage:
+		return NewGoogleCloudStorage(ctx)
+	case Filesystem:
+		return NewFilesystemStorage(ctx)
+	case None:
+		return NewNoopBlobstore(ctx)
+	default:
+		return nil, fmt.Errorf("Unknown BlobstoreType: %v", config.BlobstoreType)
+	}
+}
diff --git a/internal/util/generate.go b/internal/util/generate.go
index 5871f92..32d9cd1 100644
--- a/internal/util/generate.go
+++ b/internal/util/generate.go
@@ -22,7 +22,7 @@ import (
 	"math/big"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/exposure-notifications-server/testing/enclient"
 )
 
@@ -58,7 +58,7 @@ func RandomIntWithMin(min, max int) (int, error) {
 }
 
 func RandomTransmissionRisk() (int, error) {
-	n, err := RandomInt(database.MaxTransmissionRisk)
+	n, err := RandomInt(model.MaxTransmissionRisk)
 	return n + 1, err
 }
 
@@ -70,14 +70,14 @@ func RandomArrValue(arr []string) (string, error) {
 	return arr[n], nil
 }
 
-func GenerateExposureKeys(numKeys, tr int) []database.ExposureKey {
+func GenerateExposureKeys(numKeys, tr int) []model.ExposureKey {
 	// When publishing multiple keys - they'll be on different days.
 	intervalCount, err := RandomIntervalCount()
 	if err != nil {
 		log.Fatalf("problem with random interval: %v", err)
 	}
 	intervalNumber := int32(time.Now().Unix()/600) - intervalCount
-	exposureKeys := make([]database.ExposureKey, numKeys)
+	exposureKeys := make([]model.ExposureKey, numKeys)
 	for i := 0; i < numKeys; i++ {
 		transmissionRisk := tr
 		if transmissionRisk < 0 {
@@ -102,12 +102,12 @@ func GenerateExposureKeys(numKeys, tr int) []database.ExposureKey {
 }
 
 // Creates a random exposure key.
-func RandomExposureKey(intervalNumber enclient.Interval, intervalCount int32, transmissionRisk int) (database.ExposureKey, error) {
+func RandomExposureKey(intervalNumber enclient.Interval, intervalCount int32, transmissionRisk int) (model.ExposureKey, error) {
 	key, err := GenerateKey()
 	if err != nil {
-		return database.ExposureKey{}, err
+		return model.ExposureKey{}, err
 	}
-	return database.ExposureKey{
+	return model.ExposureKey{
 		Key:              key,
 		IntervalNumber:   int32(intervalNumber),
 		IntervalCount:    intervalCount,
diff --git a/internal/verification/verify.go b/internal/verification/verify.go
index d429b4a..9750aab 100644
--- a/internal/verification/verify.go
+++ b/internal/verification/verify.go
@@ -21,7 +21,7 @@ import (
 
 	"github.com/google/exposure-notifications-server/internal/android"
 	authorizedapp "github.com/google/exposure-notifications-server/internal/authorizedapp/model"
-	"github.com/google/exposure-notifications-server/internal/database"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
 
 	"github.com/google/exposure-notifications-server/internal/ios"
 )
@@ -33,7 +33,7 @@ var (
 
 // VerifyRegions checks the request regions against the regions allowed by
 // the configuration for the application.
-func VerifyRegions(cfg *authorizedapp.AuthorizedApp, data *database.Publish) error {
+func VerifyRegions(cfg *authorizedapp.AuthorizedApp, data *publishmodel.Publish) error {
 	if cfg == nil {
 		return fmt.Errorf("app configuration is empty")
 	}
@@ -50,7 +50,7 @@ func VerifyRegions(cfg *authorizedapp.AuthorizedApp, data *database.Publish) err
 
 // VerifySafetyNet verifies the Android SafetyNet device attestation against the
 // allowed configuration for the application.
-func VerifySafetyNet(ctx context.Context, requestTime time.Time, cfg *authorizedapp.AuthorizedApp, publish *database.Publish) error {
+func VerifySafetyNet(ctx context.Context, requestTime time.Time, cfg *authorizedapp.AuthorizedApp, publish *publishmodel.Publish) error {
 	if cfg == nil {
 		return fmt.Errorf("cannot enforce SafetyNet, missing config")
 	}
@@ -64,7 +64,7 @@ func VerifySafetyNet(ctx context.Context, requestTime time.Time, cfg *authorized
 }
 
 // VerifyDeviceCheck verifies an iOS DeviceCheck token against the Apple API.
-func VerifyDeviceCheck(ctx context.Context, cfg *authorizedapp.AuthorizedApp, data *database.Publish) error {
+func VerifyDeviceCheck(ctx context.Context, cfg *authorizedapp.AuthorizedApp, data *publishmodel.Publish) error {
 	if cfg == nil {
 		return fmt.Errorf("cannot enforce DeviceCheck, missing config")
 	}
diff --git a/terraform/main.tf b/terraform/main.tf
index 713d76b..7c5587e 100644
--- a/terraform/main.tf
+++ b/terraform/main.tf
@@ -68,6 +68,12 @@ resource "google_service_networking_connection" "private_vpc_connection" {
   reserved_peering_ranges = [google_compute_global_address.private_ip_address.name]
 }
 
+locals {
+  build_substitutions = {
+    "_REGION" : var.region,
+  }
+}
+
 # This step automatically runs a build as well, so everything that uses an image depends on it.
 resource "google_cloudbuild_trigger" "build-and-publish" {
   provider = google-beta
@@ -84,6 +90,8 @@ resource "google_cloudbuild_trigger" "build-and-publish" {
     }
   }
 
+  substitutions = local.build_substitutions
+
   depends_on = [google_project_service.services["cloudbuild.googleapis.com"]]
 }
 
diff --git a/tools/export-config/main.go b/tools/export-config/main.go
index b53c6c7..26f25c8 100644
--- a/tools/export-config/main.go
+++ b/tools/export-config/main.go
@@ -22,7 +22,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/database"
+
+	"github.com/google/exposure-notifications-server/internal/export/model"
+
 	"github.com/kelseyhightower/envconfig"
 )
 
@@ -77,30 +81,30 @@ func main() {
 	}
 
 	ctx := context.Background()
-	var config database.Config
+	var config coredb.Config
 	err := envconfig.Process("database", &config)
 	if err != nil {
 		log.Fatalf("error loading environment variables: %v", err)
 	}
 
-	db, err := database.NewFromEnv(ctx, &config)
+	db, err := coredb.NewFromEnv(ctx, &config)
 	if err != nil {
 		log.Fatalf("unable to connect to database: %v", err)
 	}
 	defer db.Close(ctx)
 
-	si := database.SignatureInfo{
+	si := model.SignatureInfo{
 		SigningKey:        *signingKey,
 		AppPackageName:    *appPkgID,
 		BundleID:          *bundleID,
 		SigningKeyVersion: *signingKeyVersion,
 		SigningKeyID:      *signingKeyID,
 	}
-	if err := db.AddSignatureInfo(ctx, &si); err != nil {
+	if err := database.New(db).AddSignatureInfo(ctx, &si); err != nil {
 		log.Fatalf("AddSignatureInfo: %v", err)
 	}
 
-	ec := database.ExportConfig{
+	ec := model.ExportConfig{
 		BucketName:       *bucketName,
 		FilenameRoot:     *filenameRoot,
 		Period:           *period,
@@ -109,7 +113,7 @@ func main() {
 		Thru:             thruTime,
 		SignatureInfoIDs: []int64{si.ID},
 	}
-	if err := db.AddExportConfig(ctx, &ec); err != nil {
+	if err := database.New(db).AddExportConfig(ctx, &ec); err != nil {
 		log.Fatalf("Failure: %v", err)
 	}
 	log.Printf("Successfully created ExportConfig %d.", ec.ConfigID)
diff --git a/tools/export-generate/main.go b/tools/export-generate/main.go
index b5e34b1..fb076ea 100644
--- a/tools/export-generate/main.go
+++ b/tools/export-generate/main.go
@@ -29,8 +29,11 @@ import (
 	"math"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/export"
+	"github.com/google/exposure-notifications-server/internal/export/model"
+
 	"github.com/google/exposure-notifications-server/internal/util"
 )
 
@@ -42,7 +45,7 @@ var (
 	region         = flag.String("region", "US", "The region for the test export.")
 	startTimestamp = flag.String("start-timestamp", "2020-05-01T15:00:00Z", "The test export start timestamp (RFC3339).")
 	endTimestamp   = flag.String("end-timestamp", "2020-05-02T15:00:00Z", "The test export end timestamp (RFC3339).")
-	numKeys        = flag.Int("num-keys", 450, "Number of total random temporary exposure keys to generate in the export")
+	numKeys        = flag.Int("num-keys", 450, "Number of total random temporary exposure keys to generate. Ignored if tek-file set.")
 	tekFile        = flag.String("tek-file", "", "JSON file of TEKs in the same format as calling publish endpoint")
 	batchSize      = flag.Int("batches-size", 100, "Max number of keys in each file in the batch")
 )
@@ -89,20 +92,20 @@ func main() {
 		log.Fatalf("problem with random transmission risk: %v", err)
 	}
 	var actualNumKeys int
-	var exposureKeys []database.Exposure
+	var exposureKeys []publishmodel.Exposure
 	if *tekFile != "" {
 		log.Printf("Using TEKs provided in: %s", *tekFile)
 		file, err := ioutil.ReadFile(*tekFile)
 		if err != nil {
 			log.Fatalf("unable to read file: %v", err)
 		}
-		data := database.ExposureKeys{}
+		data := publishmodel.ExposureKeys{}
 		err = json.Unmarshal([]byte(file), &data)
 		if err != nil {
 			log.Fatalf("unable to parse json: %v", err)
 		}
 		for _, k := range data.Keys {
-			ek, err := database.TransformExposureKey(k, "", []string{}, time.Now(), int32(0), math.MaxInt32)
+			ek, err := publishmodel.TransformExposureKey(k, "", []string{}, time.Now(), int32(0), math.MaxInt32)
 			if err != nil {
 				log.Fatalf("invalid exposure key: %v", err)
 			}
@@ -110,10 +113,11 @@ func main() {
 		}
 		actualNumKeys = len(exposureKeys)
 	} else {
+		log.Printf("Genrating %d random TEKs", *numKeys)
 		keys := util.GenerateExposureKeys(*numKeys, tr)
 		actualNumKeys = *numKeys
 
-		exposureKeys = make([]database.Exposure, actualNumKeys)
+		exposureKeys = make([]publishmodel.Exposure, actualNumKeys)
 		for i, k := range keys {
 			decoded, err := base64.StdEncoding.DecodeString(k.Key)
 			if err != nil {
@@ -130,7 +134,7 @@ func main() {
 	}
 
 	// split up into batches
-	eb := &database.ExportBatch{
+	eb := &model.ExportBatch{
 		FilenameRoot:   *filenameRoot,
 		StartTimestamp: startTime,
 		EndTimestamp:   endTime,
@@ -139,13 +143,13 @@ func main() {
 	numBatches := int(math.Ceil(float64(actualNumKeys) / float64(*batchSize)))
 	log.Printf("number of batches: %d", numBatches)
 	b := 0
-	currentBatch := []*database.Exposure{}
+	currentBatch := []*publishmodel.Exposure{}
 	for i := 0; i < actualNumKeys; i++ {
 		currentBatch = append(currentBatch, &exposureKeys[i])
 		if len(currentBatch) == *batchSize {
 			b++
 			writeFile(eb, currentBatch, b, numBatches, actualNumKeys, privateKey)
-			currentBatch = []*database.Exposure{}
+			currentBatch = []*publishmodel.Exposure{}
 		}
 	}
 	if len(currentBatch) > 0 {
@@ -154,8 +158,8 @@ func main() {
 	}
 }
 
-func writeFile(eb *database.ExportBatch, currentBatch []*database.Exposure, b, numBatches, numRecords int, privateKey *ecdsa.PrivateKey) {
-	signatureInfo := &database.SignatureInfo{
+func writeFile(eb *model.ExportBatch, currentBatch []*publishmodel.Exposure, b, numBatches, numRecords int, privateKey *ecdsa.PrivateKey) {
+	signatureInfo := &model.SignatureInfo{
 		SigningKeyID:      *keyID,
 		SigningKeyVersion: *keyVersion,
 	}
diff --git a/tools/exposure-client/main.go b/tools/exposure-client/main.go
index 32a7b49..73e5142 100644
--- a/tools/exposure-client/main.go
+++ b/tools/exposure-client/main.go
@@ -24,7 +24,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/exposure-notifications-server/internal/util"
 	"github.com/google/exposure-notifications-server/testing/enclient"
 )
@@ -84,7 +84,7 @@ func main() {
 		log.Printf("could not get random padding: %v", err)
 	}
 
-	data := database.Publish{
+	data := model.Publish{
 		Keys:           exposureKeys,
 		Regions:        region,
 		AppPackageName: *appPackage,


Test patch:
diff --git a/internal/android/safetynet_test.go b/internal/android/safetynet_test.go
index 6441744..bc8e3ff 100644
--- a/internal/android/safetynet_test.go
+++ b/internal/android/safetynet_test.go
@@ -21,7 +21,8 @@ import (
 
 	"github.com/google/exposure-notifications-server/internal/authorizedapp/model"
 	"github.com/google/exposure-notifications-server/internal/base64util"
-	"github.com/google/exposure-notifications-server/internal/database"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/go-cmp/cmp"
 )
 
@@ -39,8 +40,8 @@ var (
 	// **This comment applies to all test data in this file.**
 	payload = `eyJhbGciOiJSUzI1NiIsIng1YyI6WyJNSUlGa3pDQ0JIdWdBd0lCQWdJUkFOY1NramRzNW42K0NBQUFBQUFwYTBjd0RRWUpLb1pJaHZjTkFRRUxCUUF3UWpFTE1Ba0dBMVVFQmhNQ1ZWTXhIakFjQmdOVkJBb1RGVWR2YjJkc1pTQlVjblZ6ZENCVFpYSjJhV05sY3pFVE1CRUdBMVVFQXhNS1IxUlRJRU5CSURGUE1UQWVGdzB5TURBeE1UTXhNVFF4TkRsYUZ3MHlNVEF4TVRFeE1UUXhORGxhTUd3eEN6QUpCZ05WQkFZVEFsVlRNUk13RVFZRFZRUUlFd3BEWVd4cFptOXlibWxoTVJZd0ZBWURWUVFIRXcxTmIzVnVkR0ZwYmlCV2FXVjNNUk13RVFZRFZRUUtFd3BIYjI5bmJHVWdURXhETVJzd0dRWURWUVFERXhKaGRIUmxjM1F1WVc1a2NtOXBaQzVqYjIwd2dnRWlNQTBHQ1NxR1NJYjNEUUVCQVFVQUE0SUJEd0F3Z2dFS0FvSUJBUUNXRXJCUVRHWkdOMWlaYk45ZWhSZ2lmV0J4cWkyUGRneHcwM1A3VHlKWmZNeGpwNUw3ajFHTmVQSzVIemRyVW9JZDF5Q0l5Qk15eHFnYXpxZ3RwWDVXcHNYVzRWZk1oSmJOMVkwOXF6cXA2SkQrMlBaZG9UVTFrRlJBTVdmTC9VdVp0azdwbVJYZ0dtNWpLRHJaOU54ZTA0dk1ZUXI4OE5xd1cva2ZaMWdUT05JVVQwV3NMVC80NTIyQlJXeGZ3eGMzUUUxK1RLV2tMQ3J2ZWs2V2xJcXlhQzUyVzdNRFI4TXBGZWJ5bVNLVHZ3Zk1Sd3lLUUxUMDNVTDR2dDQ4eUVjOHNwN3dUQUhNL1dEZzhRb3RhcmY4T0JIa25vWjkyWGl2aWFWNnRRcWhST0hDZmdtbkNYaXhmVzB3RVhDdnFpTFRiUXRVYkxzUy84SVJ0ZFhrcFFCOUFnTUJBQUdqZ2dKWU1JSUNWREFPQmdOVkhROEJBZjhFQkFNQ0JhQXdFd1lEVlIwbEJBd3dDZ1lJS3dZQkJRVUhBd0V3REFZRFZSMFRBUUgvQkFJd0FEQWRCZ05WSFE0RUZnUVU2REhCd3NBdmI1M2cvQzA3cHJUdnZ3TlFRTFl3SHdZRFZSMGpCQmd3Rm9BVW1OSDRiaERyejV2c1lKOFlrQnVnNjMwSi9Tc3daQVlJS3dZQkJRVUhBUUVFV0RCV01DY0dDQ3NHQVFVRkJ6QUJoaHRvZEhSd09pOHZiMk56Y0M1d2Eya3VaMjl2Wnk5bmRITXhiekV3S3dZSUt3WUJCUVVITUFLR0gyaDBkSEE2THk5d2Eya3VaMjl2Wnk5bmMzSXlMMGRVVXpGUE1TNWpjblF3SFFZRFZSMFJCQll3RklJU1lYUjBaWE4wTG1GdVpISnZhV1F1WTI5dE1DRUdBMVVkSUFRYU1CZ3dDQVlHWjRFTUFRSUNNQXdHQ2lzR0FRUUIxbmtDQlFNd0x3WURWUjBmQkNnd0pqQWtvQ0tnSUlZZWFIUjBjRG92TDJOeWJDNXdhMmt1WjI5dlp5OUhWRk14VHpFdVkzSnNNSUlCQkFZS0t3WUJCQUhXZVFJRUFnU0I5UVNCOGdEd0FIY0E5bHlVTDlGM01DSVVWQmdJTUpSV2p1Tk5FeGt6djk4TUx5QUx6RTd4Wk9NQUFBRnZudXkwWndBQUJBTUFTREJHQWlFQTdlLzBZUnUzd0FGbVdIMjdNMnZiVmNaL21ycCs0cmZZYy81SVBKMjlGNmdDSVFDbktDQ0FhY1ZOZVlaOENDZllkR3BCMkdzSHh1TU9Ia2EvTzQxaldlRit6Z0IxQUVTVVpTNnc3czZ2eEVBSDJLaitLTURhNW9LKzJNc3h0VC9UTTVhMXRvR29BQUFCYjU3c3RKTUFBQVFEQUVZd1JBSWdFWGJpb1BiSnA5cUMwRGoyNThERkdTUk1BVStaQjFFaVZFYmJiLzRVdk5FQ0lCaEhrQnQxOHZSbjl6RHZ5cmZ4eXVkY0hUT1NsM2dUYVlBLzd5VC9CaUg0TUEwR0NTcUdTSWIzRFFFQkN3VUFBNElCQVFESUFjUUJsbWQ4TUVnTGRycnJNYkJUQ3ZwTVhzdDUrd3gyRGxmYWpKTkpVUDRqWUZqWVVROUIzWDRFMnpmNDluWDNBeXVaRnhBcU9SbmJqLzVqa1k3YThxTUowajE5ekZPQitxZXJ4ZWMwbmhtOGdZbExiUW02c0tZN1AwZXhmcjdIdUszTWtQMXBlYzE0d0ZFVWFHcUR3VWJHZ2wvb2l6MzhGWENFK0NXOEUxUUFFVWZ2YlFQVFliS3hZait0Q05sc3MwYlRTb0wyWjJkL2ozQnBMM01GdzB5eFNLL1VUcXlrTHIyQS9NZGhKUW14aStHK01LUlNzUXI2MkFuWmF1OXE2WUZvaSs5QUVIK0E0OFh0SXlzaEx5Q1RVM0h0K2FLb2hHbnhBNXVsMVhSbXFwOEh2Y0F0MzlQOTVGWkdGSmUwdXZseWpPd0F6WHVNdTdNK1BXUmMiLCJNSUlFU2pDQ0F6S2dBd0lCQWdJTkFlTzBtcUdOaXFtQkpXbFF1REFOQmdrcWhraUc5dzBCQVFzRkFEQk1NU0F3SGdZRFZRUUxFeGRIYkc5aVlXeFRhV2R1SUZKdmIzUWdRMEVnTFNCU01qRVRNQkVHQTFVRUNoTUtSMnh2WW1Gc1UybG5iakVUTUJFR0ExVUVBeE1LUjJ4dlltRnNVMmxuYmpBZUZ3MHhOekEyTVRVd01EQXdOREphRncweU1URXlNVFV3TURBd05ESmFNRUl4Q3pBSkJnTlZCQVlUQWxWVE1SNHdIQVlEVlFRS0V4VkhiMjluYkdVZ1ZISjFjM1FnVTJWeWRtbGpaWE14RXpBUkJnTlZCQU1UQ2tkVVV5QkRRU0F4VHpFd2dnRWlNQTBHQ1NxR1NJYjNEUUVCQVFVQUE0SUJEd0F3Z2dFS0FvSUJBUURRR005RjFJdk4wNXprUU85K3ROMXBJUnZKenp5T1RIVzVEekVaaEQyZVBDbnZVQTBRazI4RmdJQ2ZLcUM5RWtzQzRUMmZXQllrL2pDZkMzUjNWWk1kUy9kTjRaS0NFUFpSckF6RHNpS1VEelJybUJCSjV3dWRnem5kSU1ZY0xlL1JHR0ZsNXlPRElLZ2pFdi9TSkgvVUwrZEVhbHROMTFCbXNLK2VRbU1GKytBY3hHTmhyNTlxTS85aWw3MUkyZE44RkdmY2Rkd3VhZWo0YlhocDBMY1FCYmp4TWNJN0pQMGFNM1Q0SStEc2F4bUtGc2JqemFUTkM5dXpwRmxnT0lnN3JSMjV4b3luVXh2OHZObWtxN3pkUEdIWGt4V1k3b0c5aitKa1J5QkFCazdYckpmb3VjQlpFcUZKSlNQazdYQTBMS1cwWTN6NW96MkQwYzF0Skt3SEFnTUJBQUdqZ2dFek1JSUJMekFPQmdOVkhROEJBZjhFQkFNQ0FZWXdIUVlEVlIwbEJCWXdGQVlJS3dZQkJRVUhBd0VHQ0NzR0FRVUZCd01DTUJJR0ExVWRFd0VCL3dRSU1BWUJBZjhDQVFBd0hRWURWUjBPQkJZRUZKalIrRzRRNjgrYjdHQ2ZHSkFib090OUNmMHJNQjhHQTFVZEl3UVlNQmFBRkp2aUIxZG5IQjdBYWdiZVdiU2FMZC9jR1lZdU1EVUdDQ3NHQVFVRkJ3RUJCQ2t3SnpBbEJnZ3JCZ0VGQlFjd0FZWVphSFIwY0RvdkwyOWpjM0F1Y0d0cExtZHZiMmN2WjNOeU1qQXlCZ05WSFI4RUt6QXBNQ2VnSmFBamhpRm9kSFJ3T2k4dlkzSnNMbkJyYVM1bmIyOW5MMmR6Y2pJdlozTnlNaTVqY213d1B3WURWUjBnQkRnd05qQTBCZ1puZ1F3QkFnSXdLakFvQmdnckJnRUZCUWNDQVJZY2FIUjBjSE02THk5d2Eya3VaMjl2Wnk5eVpYQnZjMmwwYjNKNUx6QU5CZ2txaGtpRzl3MEJBUXNGQUFPQ0FRRUFHb0ErTm5uNzh5NnBSamQ5WGxRV05hN0hUZ2laL3IzUk5Ha21VbVlIUFFxNlNjdGk5UEVhanZ3UlQyaVdUSFFyMDJmZXNxT3FCWTJFVFV3Z1pRK2xsdG9ORnZoc085dHZCQ09JYXpwc3dXQzlhSjl4anU0dFdEUUg4TlZVNllaWi9YdGVEU0dVOVl6SnFQalk4cTNNRHhyem1xZXBCQ2Y1bzhtdy93SjRhMkc2eHpVcjZGYjZUOE1jRE8yMlBMUkw2dTNNNFR6czNBMk0xajZieWtKWWk4d1dJUmRBdktMV1p1L2F4QlZielltcW13a201ekxTRFc1bklBSmJFTENRQ1p3TUg1NnQyRHZxb2Z4czZCQmNDRklaVVNweHU2eDZ0ZDBWN1N2SkNDb3NpclNtSWF0ai85ZFNTVkRRaWJldDhxLzdVSzR2NFpVTjgwYXRuWnoxeWc9PSJdfQ.eyJub25jZSI6IlZreEphblUwYW5CcFkydDNObEZFZVM5d2IzRk9iRlprTHpCaFNYTktiRTVqYkVGSmFESnFRbm94VFQwPSIsInRpbWVzdGFtcE1zIjoxNTg5NTg4NDA4MTg1LCJhcGtQYWNrYWdlTmFtZSI6ImNvbS5nb29nbGUuYW5kcm9pZC5hcHBzLmV4cG9zdXJlbm90aWZpY2F0aW9uIiwiYXBrRGlnZXN0U2hhMjU2IjoiYWt2Y3oxdWgyak85NzF5blZ1bXdsSEM4L2xHZG5vYjVGVlNIWVlabCtWcz0iLCJjdHNQcm9maWxlTWF0Y2giOnRydWUsImFwa0NlcnRpZmljYXRlRGlnZXN0U2hhMjU2IjpbImpxbVlFcWk5cVV2cFVlMTFxTWYzdjJvNlZFUU0rNU5EZWUyYnoweGR6V2M9Il0sImJhc2ljSW50ZWdyaXR5Ijp0cnVlLCJldmFsdWF0aW9uVHlwZSI6IkJBU0lDIn0.cU97Z9Fax39ehQvKpa8C0I0hXIh1zclZErn5uDGdtWsVTddpf1qxSVnAwxAGhBmqbOF_YL-75dGojxT8dsGpe7LBXQr9qyRm79FjfIAAqDQrblvOxIatGbR1qkmLJITF7DAc4sAcK8CP2RxIrAkUIAUAdgwsx4JbXeNMfPxR-XODMo93MaMRLV5byA1MDaRw3RA-HTx81LEAuQJfUcoRf-O7WTnBBYDUDlUAY5_VNp7wGRKg9eNNzbbysPqpS-o-sW67dvsalav2ENwwQT9RJwWR4YCI4bDh_qH3Mz8FfChwefS1AW4nOPsOJwkEhLHIO521ZQiwZVct1b6glPc08g`
 
-	publish = &database.Publish{
-		Keys: []database.ExposureKey{
+	publish = &publishmodel.Publish{
+		Keys: []publishmodel.ExposureKey{
 			{Key: "QLIvVheW9p6JiTx4pslesg==", IntervalNumber: 2649312, IntervalCount: 144, TransmissionRisk: 1},
 			{Key: "UW7pkDKfLbfLs8LveFyY3w==", IntervalNumber: 2649168, IntervalCount: 144, TransmissionRisk: 1},
 		},
diff --git a/internal/authorizedapp/database/authorized_app_test.go b/internal/authorizedapp/database/authorized_app_test.go
index 26bc1eb..4781166 100644
--- a/internal/authorizedapp/database/authorized_app_test.go
+++ b/internal/authorizedapp/database/authorized_app_test.go
@@ -59,6 +59,7 @@ func TestMain(m *testing.M) {
 	}
 	os.Exit(m.Run())
 }
+
 func TestGetAuthorizedApp(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
@@ -239,7 +240,7 @@ func TestGetAuthorizedApp(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			config, err := NewAuthorizedAppDB(testDB).GetAuthorizedApp(ctx, sm, "myapp")
+			config, err := New(testDB).GetAuthorizedApp(ctx, sm, "myapp")
 			if (err != nil) != c.err {
 				t.Fatal(err)
 			}
diff --git a/internal/database/export_test.go b/internal/export/database/export_test.go
similarity index 74%
rename from internal/database/export_test.go
rename to internal/export/database/export_test.go
index 9a5662e..c1f1d92 100644
--- a/internal/database/export_test.go
+++ b/internal/export/database/export_test.go
@@ -17,29 +17,50 @@ package database
 import (
 	"context"
 	"errors"
+	"log"
+	"os"
 	"sort"
 	"testing"
 	"time"
 
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/export/model"
+	publishdb "github.com/google/exposure-notifications-server/internal/publish/database"
+	publishmodel "github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/go-cmp/cmp"
 	pgx "github.com/jackc/pgx/v4"
 )
 
+var testDB *database.DB
+
+func TestMain(m *testing.M) {
+	ctx := context.Background()
+
+	if os.Getenv("DB_USER") != "" {
+		var err error
+		testDB, err = database.CreateTestDB(ctx)
+		if err != nil {
+			log.Fatalf("creating test DB: %v", err)
+		}
+	}
+	os.Exit(m.Run())
+}
+
 func TestAddSignatureInfo(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	thruTime := time.Now().UTC().Add(6 * time.Hour).Truncate(time.Microsecond)
-	want := &SignatureInfo{
+	want := &model.SignatureInfo{
 		SigningKey:        "/kms/project/key/1",
 		SigningKeyVersion: "1",
 		SigningKeyID:      "310",
 		EndTimestamp:      thruTime,
 	}
-	if err := testDB.AddSignatureInfo(ctx, want); err != nil {
+	if err := New(testDB).AddSignatureInfo(ctx, want); err != nil {
 		t.Fatal(err)
 	}
 	conn, err := testDB.Pool.Acquire(ctx)
@@ -47,7 +68,7 @@ func TestAddSignatureInfo(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer conn.Release()
-	var got SignatureInfo
+	var got model.SignatureInfo
 	err = conn.QueryRow(ctx, `
 		SELECT
 		  id, signing_key, app_package_name, bundle_id, signing_key_version, signing_key_id, thru_timestamp
@@ -69,11 +90,11 @@ func TestLookupSignatureInfos(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	testTime := time.Now().UTC()
-	want := []*SignatureInfo{
+	want := []*model.SignatureInfo{
 		{
 			SigningKey:        "/kms/project/key/version/1",
 			SigningKeyVersion: "1",
@@ -93,11 +114,11 @@ func TestLookupSignatureInfos(t *testing.T) {
 		},
 	}
 	for _, si := range want {
-		testDB.AddSignatureInfo(ctx, si)
+		New(testDB).AddSignatureInfo(ctx, si)
 	}
 
 	ids := []int64{want[0].ID, want[1].ID, want[2].ID}
-	got, err := testDB.LookupSignatureInfos(ctx, ids, testTime)
+	got, err := New(testDB).LookupSignatureInfos(ctx, ids, testTime)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -114,12 +135,12 @@ func TestAddExportConfig(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	fromTime := time.Now()
 	thruTime := fromTime.Add(6 * time.Hour)
-	want := &ExportConfig{
+	want := &model.ExportConfig{
 		BucketName:       "mocked",
 		FilenameRoot:     "root",
 		Period:           3 * time.Hour,
@@ -128,7 +149,7 @@ func TestAddExportConfig(t *testing.T) {
 		Thru:             thruTime,
 		SignatureInfoIDs: []int64{42, 84},
 	}
-	if err := testDB.AddExportConfig(ctx, want); err != nil {
+	if err := New(testDB).AddExportConfig(ctx, want); err != nil {
 		t.Fatal(err)
 	}
 	conn, err := testDB.Pool.Acquire(ctx)
@@ -137,7 +158,7 @@ func TestAddExportConfig(t *testing.T) {
 	}
 	defer conn.Release()
 	var (
-		got   ExportConfig
+		got   model.ExportConfig
 		psecs int
 	)
 	err = conn.QueryRow(ctx, `
@@ -164,11 +185,11 @@ func TestIterateExportConfigs(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	now := time.Now().Truncate(time.Microsecond)
-	ecs := []*ExportConfig{
+	ecs := []*model.ExportConfig{
 		{
 			BucketName:   "b 1",
 			FilenameRoot: "active 1",
@@ -196,13 +217,13 @@ func TestIterateExportConfigs(t *testing.T) {
 	for _, ec := range ecs {
 		ec.Period = time.Hour
 		ec.Region = "R"
-		if err := testDB.AddExportConfig(ctx, ec); err != nil {
+		if err := New(testDB).AddExportConfig(ctx, ec); err != nil {
 			t.Fatal(err)
 		}
 	}
 
-	var got []*ExportConfig
-	err := testDB.IterateExportConfigs(ctx, now, func(m *ExportConfig) error {
+	var got []*model.ExportConfig
+	err := New(testDB).IterateExportConfigs(ctx, now, func(m *model.ExportConfig) error {
 		got = append(got, m)
 		return nil
 	})
@@ -220,11 +241,11 @@ func TestBatches(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	now := time.Now().Truncate(time.Microsecond)
-	config := &ExportConfig{
+	config := &model.ExportConfig{
 		BucketName:       "mocked",
 		FilenameRoot:     "root",
 		Period:           time.Hour,
@@ -233,31 +254,31 @@ func TestBatches(t *testing.T) {
 		Thru:             now.Add(time.Hour),
 		SignatureInfoIDs: []int64{1, 2, 3, 4},
 	}
-	if err := testDB.AddExportConfig(ctx, config); err != nil {
+	if err := New(testDB).AddExportConfig(ctx, config); err != nil {
 		t.Fatal(err)
 	}
-	var batches []*ExportBatch
+	var batches []*model.ExportBatch
 	var wantLatest time.Time
 	for i := 0; i < 4; i++ {
 		start := now.Add(time.Duration(i) * time.Minute)
 		end := start.Add(time.Minute)
 		wantLatest = end
-		batches = append(batches, &ExportBatch{
+		batches = append(batches, &model.ExportBatch{
 			ConfigID:         config.ConfigID,
 			BucketName:       config.BucketName,
 			FilenameRoot:     config.FilenameRoot,
 			Region:           config.Region,
-			Status:           ExportBatchOpen,
+			Status:           model.ExportBatchOpen,
 			StartTimestamp:   start,
 			EndTimestamp:     end,
 			SignatureInfoIDs: []int64{1, 2, 3, 4},
 		})
 	}
-	if err := testDB.AddExportBatches(ctx, batches); err != nil {
+	if err := New(testDB).AddExportBatches(ctx, batches); err != nil {
 		t.Fatal(err)
 	}
 
-	gotLatest, err := testDB.LatestExportBatchEnd(ctx, config)
+	gotLatest, err := New(testDB).LatestExportBatchEnd(ctx, config)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -270,7 +291,7 @@ func TestBatches(t *testing.T) {
 		var batchID int64
 		// Lease all the batches.
 		for range batches {
-			got, err := testDB.LeaseBatch(ctx, time.Hour, now)
+			got, err := New(testDB).LeaseBatch(ctx, time.Hour, now)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -283,7 +304,7 @@ func TestBatches(t *testing.T) {
 					got.ConfigID, got.BucketName, got.FilenameRoot, got.Region,
 					config.ConfigID, config.BucketName, config.FilenameRoot, config.Region)
 			}
-			if got.Status != ExportBatchPending {
+			if got.Status != model.ExportBatchPending {
 				t.Errorf("LeaseBatch: got status %q, want pending", got.Status)
 			}
 			wantExpires := now.Add(time.Hour)
@@ -293,14 +314,14 @@ func TestBatches(t *testing.T) {
 			batchID = got.BatchID
 		}
 		// Every batch is leased.
-		got, err := testDB.LeaseBatch(ctx, time.Hour, now)
+		got, err := New(testDB).LeaseBatch(ctx, time.Hour, now)
 		if got != nil || err != nil {
 			t.Errorf("all leased: got (%v, %v), want (nil, nil)", got, err)
 		}
 		return batchID
 	}
 	// Now, all end times are in the future, so no batches can be leased.
-	got, err := testDB.LeaseBatch(ctx, time.Hour, now)
+	got, err := New(testDB).LeaseBatch(ctx, time.Hour, now)
 	if got != nil || err != nil {
 		t.Errorf("got (%v, %v), want (nil, nil)", got, err)
 	}
@@ -313,15 +334,15 @@ func TestBatches(t *testing.T) {
 	batchID := leaseBatches()
 
 	// Complete a batch.
-	err = testDB.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error { return completeBatch(ctx, tx, batchID) })
+	err = testDB.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error { return completeBatch(ctx, tx, batchID) })
 	if err != nil {
 		t.Fatal(err)
 	}
-	got, err = testDB.LookupExportBatch(ctx, batchID)
+	got, err = New(testDB).LookupExportBatch(ctx, batchID)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if got.Status != ExportBatchComplete {
+	if got.Status != model.ExportBatchComplete {
 		t.Errorf("after completion: got status %q, want complete", got.Status)
 	}
 }
@@ -330,68 +351,69 @@ func TestFinalizeBatch(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	exportDB := New(testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 	now := time.Now().Truncate(time.Microsecond)
 
 	// Add a config.
-	ec := &ExportConfig{
+	ec := &model.ExportConfig{
 		BucketName:   "some-bucket",
 		FilenameRoot: "filename-root",
 		Period:       time.Minute,
 		Region:       "US",
 	}
-	if err := testDB.AddExportConfig(ctx, ec); err != nil {
+	if err := exportDB.AddExportConfig(ctx, ec); err != nil {
 		t.Fatal(err)
 	}
 
 	// Add a batch.
-	eb := &ExportBatch{
+	eb := &model.ExportBatch{
 		ConfigID:       ec.ConfigID,
 		BucketName:     ec.BucketName,
 		FilenameRoot:   ec.FilenameRoot,
 		StartTimestamp: now.Add(-2 * time.Hour),
 		EndTimestamp:   now.Add(-time.Hour),
 		Region:         ec.Region,
-		Status:         ExportBatchOpen,
+		Status:         model.ExportBatchOpen,
 	}
-	if err := testDB.AddExportBatches(ctx, []*ExportBatch{eb}); err != nil {
+	if err := exportDB.AddExportBatches(ctx, []*model.ExportBatch{eb}); err != nil {
 		t.Fatal(err)
 	}
 
 	// Lease the batch.
-	eb, err := testDB.LeaseBatch(ctx, time.Hour, now)
+	eb, err := exportDB.LeaseBatch(ctx, time.Hour, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Check that the batch is PENDING.
-	gotBatch, err := testDB.LookupExportBatch(ctx, eb.BatchID)
+	gotBatch, err := exportDB.LookupExportBatch(ctx, eb.BatchID)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if gotBatch.Status != ExportBatchPending {
-		t.Errorf("pre gotBatch.Status=%q, want=%q", gotBatch.Status, ExportBatchPending)
+	if gotBatch.Status != model.ExportBatchPending {
+		t.Errorf("pre gotBatch.Status=%q, want=%q", gotBatch.Status, model.ExportBatchPending)
 	}
 
 	// Finalize the batch.
 	files := []string{"file1.txt", "file2.txt"}
 	batchSize := 10
-	if err := testDB.FinalizeBatch(ctx, eb, files, batchSize); err != nil {
+	if err := exportDB.FinalizeBatch(ctx, eb, files, batchSize); err != nil {
 		t.Fatal(err)
 	}
 
 	// Check that the batch is COMPLETED.
-	gotBatch, err = testDB.LookupExportBatch(ctx, eb.BatchID)
+	gotBatch, err = exportDB.LookupExportBatch(ctx, eb.BatchID)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if gotBatch.Status != ExportBatchComplete {
-		t.Errorf("post gotBatch.Status=%q, want=%q", gotBatch.Status, ExportBatchComplete)
+	if gotBatch.Status != model.ExportBatchComplete {
+		t.Errorf("post gotBatch.Status=%q, want=%q", gotBatch.Status, model.ExportBatchComplete)
 	}
 
 	// Check that files were written.
-	gotFiles, err := testDB.LookupExportFiles(ctx, eb.ConfigID)
+	gotFiles, err := exportDB.LookupExportFiles(ctx, eb.ConfigID)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -400,18 +422,18 @@ func TestFinalizeBatch(t *testing.T) {
 	}
 
 	for i, filename := range gotFiles {
-		got, err := testDB.LookupExportFile(ctx, filename)
+		got, err := exportDB.LookupExportFile(ctx, filename)
 		if err != nil {
 			t.Fatal(err)
 		}
-		want := &ExportFile{
+		want := &model.ExportFile{
 			BucketName: eb.BucketName,
 			Filename:   filename,
 			BatchID:    eb.BatchID,
 			Region:     eb.Region,
 			BatchNum:   i + 1,
 			BatchSize:  batchSize,
-			Status:     ExportBatchComplete,
+			Status:     model.ExportBatchComplete,
 		}
 		if diff := cmp.Diff(want, got); diff != "" {
 			t.Errorf("mismatch for %q (-want, +got):\n%s", filename, diff)
@@ -424,60 +446,60 @@ func TestKeysInBatch(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
 	ctx := context.Background()
 	now := time.Now()
 
 	// Add a config.
-	ec := &ExportConfig{
+	ec := &model.ExportConfig{
 		BucketName:   "bucket-name",
 		FilenameRoot: "filename-root",
 		Period:       3600 * time.Second,
 		Region:       "US",
 		From:         now.Add(-24 * time.Hour),
 	}
-	if err := testDB.AddExportConfig(ctx, ec); err != nil {
+	if err := New(testDB).AddExportConfig(ctx, ec); err != nil {
 		t.Fatal(err)
 	}
 
 	// Create a batch for two hours ago to one hour ago.
 	startTimestamp := now.Truncate(time.Hour).Add(-2 * time.Hour)
 	endTimestamp := startTimestamp.Add(time.Hour)
-	eb := &ExportBatch{
+	eb := &model.ExportBatch{
 		ConfigID:       ec.ConfigID,
 		BucketName:     ec.BucketName,
 		FilenameRoot:   ec.FilenameRoot,
 		StartTimestamp: startTimestamp,
 		EndTimestamp:   endTimestamp,
 		Region:         ec.Region,
-		Status:         ExportBatchOpen,
+		Status:         model.ExportBatchOpen,
 	}
-	if err := testDB.AddExportBatches(ctx, []*ExportBatch{eb}); err != nil {
+	if err := New(testDB).AddExportBatches(ctx, []*model.ExportBatch{eb}); err != nil {
 		t.Fatal(err)
 	}
 
 	// Create key aligned with the StartTimestamp
-	sek := &Exposure{
+	sek := &publishmodel.Exposure{
 		ExposureKey: []byte("aaa"),
 		Regions:     []string{ec.Region},
 		CreatedAt:   startTimestamp,
 	}
 
 	// Create key aligned with the EndTimestamp
-	eek := &Exposure{
+	eek := &publishmodel.Exposure{
 		ExposureKey: []byte("bbb"),
 		Regions:     []string{ec.Region},
 		CreatedAt:   endTimestamp,
 	}
 
 	// Add the keys to the database.
-	if err := testDB.InsertExposures(ctx, []*Exposure{sek, eek}); err != nil {
+	if err := publishdb.New(testDB).InsertExposures(ctx, []*publishmodel.Exposure{sek, eek}); err != nil {
 		t.Fatal(err)
 	}
 
 	// Re-fetch the ExposureBatch by leasing it; this is important to this test which is trying
 	// to ensure our dates are going in-and-out of the database correctly.
-	leased, err := testDB.LeaseBatch(ctx, time.Hour, now)
+	leased, err := New(testDB).LeaseBatch(ctx, time.Hour, now)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -492,14 +514,14 @@ func TestKeysInBatch(t *testing.T) {
 
 	// Lookup the keys; they must be only the key created_at the startTimestamp
 	// (because start is inclusive, end is exclusive).
-	criteria := IterateExposuresCriteria{
+	criteria := publishdb.IterateExposuresCriteria{
 		IncludeRegions: []string{leased.Region},
 		SinceTimestamp: leased.StartTimestamp,
 		UntilTimestamp: leased.EndTimestamp,
 	}
 
-	var got []*Exposure
-	_, err = testDB.IterateExposures(ctx, criteria, func(exp *Exposure) error {
+	var got []*publishmodel.Exposure
+	_, err = publishdb.New(testDB).IterateExposures(ctx, criteria, func(exp *publishmodel.Exposure) error {
 		got = append(got, exp)
 		return nil
 	})
@@ -521,33 +543,34 @@ func TestAddExportFileSkipsDuplicates(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer database.ResetTestDB(t, testDB)
+	exportDB := New(testDB)
 	ctx := context.Background()
 
 	// Add foreign key records.
-	ec := &ExportConfig{Period: time.Hour}
-	if err := testDB.AddExportConfig(ctx, ec); err != nil {
+	ec := &model.ExportConfig{Period: time.Hour}
+	if err := exportDB.AddExportConfig(ctx, ec); err != nil {
 		t.Fatal(err)
 	}
-	eb := &ExportBatch{ConfigID: ec.ConfigID, Status: ExportBatchOpen}
-	if err := testDB.AddExportBatches(ctx, []*ExportBatch{eb}); err != nil {
+	eb := &model.ExportBatch{ConfigID: ec.ConfigID, Status: model.ExportBatchOpen}
+	if err := exportDB.AddExportBatches(ctx, []*model.ExportBatch{eb}); err != nil {
 		t.Fatal(err)
 	}
 	// Lease the batch to get the ID.
-	eb, err := testDB.LeaseBatch(ctx, time.Hour, time.Now())
+	eb, err := exportDB.LeaseBatch(ctx, time.Hour, time.Now())
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	wantBucketName := "bucket-1"
-	ef := &ExportFile{
+	ef := &model.ExportFile{
 		Filename:   "file",
 		BucketName: wantBucketName,
 		BatchID:    eb.BatchID,
 	}
 
 	// Add a record.
-	err = testDB.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	err = testDB.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		if err := addExportFile(ctx, tx, ef); err != nil {
 			return err
 		}
@@ -558,7 +581,7 @@ func TestAddExportFileSkipsDuplicates(t *testing.T) {
 	}
 
 	// Check that the row is present.
-	got, err := testDB.LookupExportFile(ctx, ef.Filename)
+	got, err := exportDB.LookupExportFile(ctx, ef.Filename)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -568,9 +591,9 @@ func TestAddExportFileSkipsDuplicates(t *testing.T) {
 
 	// Add a second record with same filename, must return ErrKeyConflict, and not overwrite.
 	ef.BucketName = "bucket-2"
-	err = testDB.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+	err = testDB.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
 		if err := addExportFile(ctx, tx, ef); err != nil {
-			if err == ErrKeyConflict {
+			if err == database.ErrKeyConflict {
 				return nil // Expected result.
 			}
 			return err
@@ -582,7 +605,7 @@ func TestAddExportFileSkipsDuplicates(t *testing.T) {
 	}
 
 	// Row must not be updated.
-	got, err = testDB.LookupExportFile(ctx, ef.Filename)
+	got, err = exportDB.LookupExportFile(ctx, ef.Filename)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/internal/export/worker_test.go b/internal/export/worker_test.go
index 4407d3b..4755504 100644
--- a/internal/export/worker_test.go
+++ b/internal/export/worker_test.go
@@ -19,24 +19,24 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 )
 
 func TestRandomInt(t *testing.T) {
 	expected := make(map[int]bool)
-	for i := database.MinTransmissionRisk; i <= database.MaxTransmissionRisk; i++ {
+	for i := model.MinTransmissionRisk; i <= model.MaxTransmissionRisk; i++ {
 		expected[i] = true
 	}
 
 	// Run through 1,000 iterations. To ensure the entire range can be hit.
 	for i := 0; i < 1000; i++ {
-		v, err := randomInt(database.MinTransmissionRisk, database.MaxTransmissionRisk)
+		v, err := randomInt(model.MinTransmissionRisk, model.MaxTransmissionRisk)
 		if err != nil {
 			t.Fatalf("error getting random data")
 		}
-		if v < database.MinTransmissionRisk || v > database.MaxTransmissionRisk {
+		if v < model.MinTransmissionRisk || v > model.MaxTransmissionRisk {
 			t.Fatalf("random data outside expected bounds. %v <= %v <= %v",
-				database.MinTransmissionRisk, v, database.MaxTransmissionRisk)
+				model.MinTransmissionRisk, v, model.MaxTransmissionRisk)
 		}
 		delete(expected, v)
 	}
@@ -46,7 +46,7 @@ func TestRandomInt(t *testing.T) {
 }
 
 func TestDoNotPadZeroLength(t *testing.T) {
-	exposures := make([]*database.Exposure, 0)
+	exposures := make([]*model.Exposure, 0)
 	exposures, err := ensureMinNumExposures(exposures, "US", 1000, 100)
 	if err != nil {
 		t.Fatalf("unepected error: %v", err)
@@ -56,14 +56,14 @@ func TestDoNotPadZeroLength(t *testing.T) {
 	}
 }
 
-func addExposure(t *testing.T, exposures []*database.Exposure, interval, count int32, risk int) []*database.Exposure {
-	key := make([]byte, database.KeyLength)
+func addExposure(t *testing.T, exposures []*model.Exposure, interval, count int32, risk int) []*model.Exposure {
+	key := make([]byte, model.KeyLength)
 	_, err := rand.Read(key)
 	if err != nil {
 		t.Fatalf("error getting random data: %v", err)
 	}
 	return append(exposures,
-		&database.Exposure{
+		&model.Exposure{
 			ExposureKey:      key,
 			TransmissionRisk: risk,
 			IntervalNumber:   interval,
@@ -73,11 +73,11 @@ func addExposure(t *testing.T, exposures []*database.Exposure, interval, count i
 
 func TestEnsureMinExposures(t *testing.T) {
 	expectedTR := make(map[int]bool)
-	for i := database.MinTransmissionRisk; i <= database.MaxTransmissionRisk; i++ {
+	for i := model.MinTransmissionRisk; i <= model.MaxTransmissionRisk; i++ {
 		expectedTR[i] = true
 	}
 	// Insert a few exposures - that will be used to base the interval information off of.
-	exposures := make([]*database.Exposure, 0)
+	exposures := make([]*model.Exposure, 0)
 	exposures = addExposure(t, exposures, 123456, 144, 0)
 	exposures = addExposure(t, exposures, 789101, 88, 0)
 	// all of these combinations are expected
diff --git a/internal/federationin/federationin_test.go b/internal/federationin/federationin_test.go
index 9fd0ffa..8daef98 100644
--- a/internal/federationin/federationin_test.go
+++ b/internal/federationin/federationin_test.go
@@ -20,6 +20,8 @@ import (
 	"time"
 
 	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/metrics"
 	"github.com/google/exposure-notifications-server/internal/pb"
 
@@ -38,7 +40,7 @@ var (
 )
 
 // makeRemoteExposure returns a mock model.Exposure with LocalProvenance=false.
-func makeRemoteExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *database.Exposure {
+func makeRemoteExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *model.Exposure {
 	inf := makeExposure(diagKey, diagStatus, regions...)
 	inf.LocalProvenance = false
 	inf.FederationSyncID = syncID
@@ -62,12 +64,12 @@ func (r *remoteFetchServer) fetch(ctx context.Context, req *pb.FederationFetchRe
 	return response, nil
 }
 
-// exposureDB mocks the database, recording exposure insertions.
-type exposureDB struct {
-	exposures []*database.Exposure
+// publishDB mocks the database, recording exposure insertions.
+type publishDB struct {
+	exposures []*model.Exposure
 }
 
-func (idb *exposureDB) insertExposures(ctx context.Context, exposures []*database.Exposure) error {
+func (idb *publishDB) insertExposures(ctx context.Context, exposures []*model.Exposure) error {
 	idb.exposures = append(idb.exposures, exposures...)
 	return nil
 }
@@ -99,7 +101,7 @@ func TestFederationPull(t *testing.T) {
 		name             string
 		batchSize        int
 		fetchResponses   []*pb.FederationFetchResponse
-		wantExposures    []*database.Exposure
+		wantExposures    []*model.Exposure
 		wantTokens       []string
 		wantMaxTimestamp time.Time
 	}{
@@ -135,7 +137,7 @@ func TestFederationPull(t *testing.T) {
 					FetchResponseKeyTimestamp: 400,
 				},
 			},
-			wantExposures: []*database.Exposure{
+			wantExposures: []*model.Exposure{
 				makeRemoteExposure(aaa, 1, "US"),
 				makeRemoteExposure(bbb, 1, "US"),
 				makeRemoteExposure(ccc, 2, "CA", "US"),
@@ -171,7 +173,7 @@ func TestFederationPull(t *testing.T) {
 					FetchResponseKeyTimestamp: 400,
 				},
 			},
-			wantExposures: []*database.Exposure{
+			wantExposures: []*model.Exposure{
 				makeRemoteExposure(aaa, 1, "US"),
 				makeRemoteExposure(bbb, 1, "US"),
 			},
@@ -212,7 +214,7 @@ func TestFederationPull(t *testing.T) {
 					FetchResponseKeyTimestamp: 400,
 				},
 			},
-			wantExposures: []*database.Exposure{
+			wantExposures: []*model.Exposure{
 				makeRemoteExposure(aaa, 8, "US"),
 				makeRemoteExposure(bbb, 8, "US"),
 				makeRemoteExposure(ccc, 7, "US"),
@@ -249,7 +251,7 @@ func TestFederationPull(t *testing.T) {
 					FetchResponseKeyTimestamp: 400,
 				},
 			},
-			wantExposures: []*database.Exposure{
+			wantExposures: []*model.Exposure{
 				makeRemoteExposure(aaa, 3, "US"),
 				makeRemoteExposure(bbb, 3, "US"),
 				makeRemoteExposure(ccc, 2, "CA", "US"),
@@ -265,7 +267,7 @@ func TestFederationPull(t *testing.T) {
 			ctx := context.Background()
 			query := &database.FederationInQuery{}
 			remote := remoteFetchServer{responses: tc.fetchResponses}
-			idb := exposureDB{}
+			idb := publishDB{}
 			sdb := syncDB{}
 			batchStart := time.Now()
 			if tc.batchSize > 0 {
@@ -284,7 +286,7 @@ func TestFederationPull(t *testing.T) {
 				t.Fatalf("pull returned err=%v, want err=nil", err)
 			}
 
-			if diff := cmp.Diff(tc.wantExposures, idb.exposures, cmpopts.IgnoreFields(database.Exposure{}, "CreatedAt")); diff != "" {
+			if diff := cmp.Diff(tc.wantExposures, idb.exposures, cmpopts.IgnoreFields(model.Exposure{}, "CreatedAt")); diff != "" {
 				t.Errorf("exposures mismatch (-want +got):\n%s", diff)
 			}
 			if diff := cmp.Diff(tc.wantTokens, remote.gotTokens); diff != "" {
@@ -309,8 +311,8 @@ func TestFederationPull(t *testing.T) {
 	}
 }
 
-func makeExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *database.Exposure {
-	return &database.Exposure{
+func makeExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *model.Exposure {
+	return &model.Exposure{
 		Regions:          regions,
 		TransmissionRisk: diagStatus,
 		ExposureKey:      diagKey.ExposureKey,
diff --git a/internal/federationout/federationout_test.go b/internal/federationout/federationout_test.go
index 57cea09..97660df 100644
--- a/internal/federationout/federationout_test.go
+++ b/internal/federationout/federationout_test.go
@@ -20,7 +20,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/database"
+
+	"github.com/google/exposure-notifications-server/internal/publish/model"
+
 	"github.com/google/exposure-notifications-server/internal/pb"
 	"github.com/google/exposure-notifications-server/internal/serverenv"
 	"google.golang.org/grpc/metadata"
@@ -44,9 +47,9 @@ var (
 	ddd = &pb.ExposureKey{ExposureKey: []byte("ddd"), IntervalNumber: 4}
 )
 
-// makeExposure returns a mock database.Exposure.
-func makeExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *database.Exposure {
-	return &database.Exposure{
+// makeExposure returns a mock model.Exposure.
+func makeExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *model.Exposure {
+	return &model.Exposure{
 		Regions:          regions,
 		TransmissionRisk: diagStatus,
 		ExposureKey:      diagKey.ExposureKey,
@@ -60,11 +63,11 @@ func makeExposure(diagKey *pb.ExposureKey, diagStatus int, regions ...string) *d
 type timeout struct{}
 
 func iterFunc(elements []interface{}) iterateExposuresFunc {
-	return func(_ context.Context, _ database.IterateExposuresCriteria, f func(*database.Exposure) error) (string, error) {
+	return func(_ context.Context, _ database.IterateExposuresCriteria, f func(*model.Exposure) error) (string, error) {
 		var cursor string
 		for _, el := range elements {
 			switch v := el.(type) {
-			case *database.Exposure:
+			case *model.Exposure:
 				// Set the cursor to the most recent diagnosis key, suffixed with "_cursor".
 				cursor = string(v.ExposureKey) + "_cursor"
 				if err := f(v); err != nil {
diff --git a/internal/jsonutil/unmarshal_test.go b/internal/jsonutil/unmarshal_test.go
index 1a410ec..08a77c3 100644
--- a/internal/jsonutil/unmarshal_test.go
+++ b/internal/jsonutil/unmarshal_test.go
@@ -23,7 +23,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/go-cmp/cmp"
 )
 
@@ -33,7 +33,7 @@ func TestInvalidHeader(t *testing.T) {
 	r.Header.Set("content-type", "application/text")
 
 	w := httptest.NewRecorder()
-	data := &database.Publish{}
+	data := &model.Publish{}
 	code, err := Unmarshal(w, r, data)
 
 	expCode := http.StatusUnsupportedMediaType
@@ -130,7 +130,7 @@ func TestValidPublishMessage(t *testing.T) {
 
 	w := httptest.NewRecorder()
 
-	got := &database.Publish{}
+	got := &model.Publish{}
 	code, err := Unmarshal(w, r, got)
 	if err != nil {
 		t.Fatalf("unexpected err, %v", err)
@@ -139,8 +139,8 @@ func TestValidPublishMessage(t *testing.T) {
 		t.Errorf("unmarshal wanted %v response code, got %v", http.StatusOK, code)
 	}
 
-	want := &database.Publish{
-		Keys: []database.ExposureKey{
+	want := &model.Publish{
+		Keys: []model.ExposureKey{
 			{Key: "ABC", IntervalNumber: intervalNumber, IntervalCount: 144, TransmissionRisk: 2},
 			{Key: "DEF", IntervalNumber: intervalNumber, IntervalCount: 122, TransmissionRisk: 2},
 			{Key: "123", IntervalNumber: intervalNumber, IntervalCount: 1, TransmissionRisk: 2},
@@ -161,10 +161,10 @@ func unmarshalTestHelper(t *testing.T, payloads []string, errors []string, expCo
 	for i, testStr := range payloads {
 		body := ioutil.NopCloser(bytes.NewReader([]byte(testStr)))
 		r := httptest.NewRequest("POST", "/", body)
-		r.Header.Set("content-type", "application/json")
+		r.Header.Set("content-type", "application/json; charset=utf-8")
 
 		w := httptest.NewRecorder()
-		data := &database.Publish{}
+		data := &model.Publish{}
 		code, err := Unmarshal(w, r, data)
 		if code != expCode {
 			t.Errorf("unmarshal wanted %v response code, got %v", expCode, code)
diff --git a/internal/database/exposure_test.go b/internal/publish/database/exposure_test.go
similarity index 78%
rename from internal/database/exposure_test.go
rename to internal/publish/database/exposure_test.go
index d737288..329bb99 100644
--- a/internal/database/exposure_test.go
+++ b/internal/publish/database/exposure_test.go
@@ -17,22 +17,43 @@ package database
 import (
 	"context"
 	"errors"
+	"log"
+	"os"
 	"testing"
 	"time"
 
+	coredb "github.com/google/exposure-notifications-server/internal/database"
+
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 	"github.com/google/go-cmp/cmp"
 )
 
+var testDB *coredb.DB
+var testPublishDB *PublishDB
+
+func TestMain(m *testing.M) {
+	ctx := context.Background()
+
+	if os.Getenv("DB_USER") != "" {
+		var err error
+		testDB, err = coredb.CreateTestDB(ctx)
+		if err != nil {
+			log.Fatalf("creating test DB: %v", err)
+		}
+		testPublishDB = New(testDB)
+	}
+	os.Exit(m.Run())
+}
 func TestExposures(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer coredb.ResetTestDB(t, testDB)
 	ctx := context.Background()
 
 	// Insert some Exposures.
 	batchTime := time.Date(2020, 5, 1, 0, 0, 0, 0, time.UTC).Truncate(time.Microsecond)
-	exposures := []*Exposure{
+	exposures := []*model.Exposure{
 		{
 			ExposureKey:     []byte("ABC"),
 			Regions:         []string{"US", "CA", "MX"},
@@ -66,7 +87,7 @@ func TestExposures(t *testing.T) {
 			LocalProvenance: false,
 		},
 	}
-	if err := testDB.InsertExposures(ctx, exposures); err != nil {
+	if err := testPublishDB.InsertExposures(ctx, exposures); err != nil {
 		t.Fatal(err)
 	}
 
@@ -112,7 +133,7 @@ func TestExposures(t *testing.T) {
 		if err != nil {
 			t.Fatalf("%+v: %v", test.criteria, err)
 		}
-		var want []*Exposure
+		var want []*model.Exposure
 		for _, i := range test.want {
 			want = append(want, exposures[i])
 		}
@@ -122,7 +143,7 @@ func TestExposures(t *testing.T) {
 	}
 
 	// Delete some exposures.
-	gotN, err := testDB.DeleteExposures(ctx, exposures[2].CreatedAt)
+	gotN, err := testPublishDB.DeleteExposures(ctx, exposures[2].CreatedAt)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -140,9 +161,9 @@ func TestExposures(t *testing.T) {
 
 }
 
-func listExposures(ctx context.Context, c IterateExposuresCriteria) (_ []*Exposure, err error) {
-	var exps []*Exposure
-	_, err = testDB.IterateExposures(ctx, c, func(e *Exposure) error {
+func listExposures(ctx context.Context, c IterateExposuresCriteria) (_ []*model.Exposure, err error) {
+	var exps []*model.Exposure
+	_, err = testPublishDB.IterateExposures(ctx, c, func(e *model.Exposure) error {
 		exps = append(exps, e)
 		return nil
 	})
@@ -156,10 +177,10 @@ func TestIterateExposuresCursor(t *testing.T) {
 	if testDB == nil {
 		t.Skip("no test DB")
 	}
-	defer ResetTestDB(t, testDB)
+	defer coredb.ResetTestDB(t, testDB)
 	ctx, cancel := context.WithCancel(context.Background())
 	// Insert some Exposures.
-	exposures := []*Exposure{
+	exposures := []*model.Exposure{
 		{
 			ExposureKey:    []byte("ABC"),
 			Regions:        []string{"US", "CA", "MX"},
@@ -176,12 +197,12 @@ func TestIterateExposuresCursor(t *testing.T) {
 			Regions:        []string{"MX", "CA"},
 		},
 	}
-	if err := testDB.InsertExposures(ctx, exposures); err != nil {
+	if err := testPublishDB.InsertExposures(ctx, exposures); err != nil {
 		t.Fatal(err)
 	}
 	// Iterate over them, canceling the context in the middle.
-	var seen []*Exposure
-	cursor, err := testDB.IterateExposures(ctx, IterateExposuresCriteria{}, func(e *Exposure) error {
+	var seen []*model.Exposure
+	cursor, err := testPublishDB.IterateExposures(ctx, IterateExposuresCriteria{}, func(e *model.Exposure) error {
 		seen = append(seen, e)
 		if len(seen) == 2 {
 			cancel()
@@ -200,8 +221,8 @@ func TestIterateExposuresCursor(t *testing.T) {
 	// Resume from the cursor.
 	ctx = context.Background()
 	seen = nil
-	cursor, err = testDB.IterateExposures(ctx, IterateExposuresCriteria{LastCursor: cursor},
-		func(e *Exposure) error { seen = append(seen, e); return nil })
+	cursor, err = testPublishDB.IterateExposures(ctx, IterateExposuresCriteria{LastCursor: cursor},
+		func(e *model.Exposure) error { seen = append(seen, e); return nil })
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/internal/database/exposure_model_test.go b/internal/publish/model/exposure_model_test.go
similarity index 99%
rename from internal/database/exposure_model_test.go
rename to internal/publish/model/exposure_model_test.go
index 63630e7..58e72ac 100644
--- a/internal/database/exposure_model_test.go
+++ b/internal/publish/model/exposure_model_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package database
+package model
 
 import (
 	"crypto/rand"
diff --git a/internal/verification/verify_test.go b/internal/verification/verify_test.go
index 2ce3c5e..b87de9c 100644
--- a/internal/verification/verify_test.go
+++ b/internal/verification/verify_test.go
@@ -23,7 +23,7 @@ import (
 
 	"github.com/google/exposure-notifications-server/internal/android"
 	authorizedapp "github.com/google/exposure-notifications-server/internal/authorizedapp/model"
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 )
 
 const (
@@ -33,26 +33,26 @@ const (
 func TestVerifyRegions(t *testing.T) {
 	cases := []struct {
 		name string
-		data *database.Publish
+		data *model.Publish
 		cfg  *authorizedapp.AuthorizedApp
 		err  bool
 	}{
 		{
 			name: "nil_config",
-			data: &database.Publish{Regions: []string{"US"}},
+			data: &model.Publish{Regions: []string{"US"}},
 			cfg:  nil,
 			err:  true,
 		},
 		{
 			name: "nil_regions_allows_all",
-			data: &database.Publish{Regions: []string{"US"}},
+			data: &model.Publish{Regions: []string{"US"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 			},
 		},
 		{
 			name: "empty_regions_allows_all",
-			data: &database.Publish{Regions: []string{"US"}},
+			data: &model.Publish{Regions: []string{"US"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 				AllowedRegions: map[string]struct{}{},
@@ -60,7 +60,7 @@ func TestVerifyRegions(t *testing.T) {
 		},
 		{
 			name: "region_matches_one",
-			data: &database.Publish{Regions: []string{"US"}},
+			data: &model.Publish{Regions: []string{"US"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 				AllowedRegions: map[string]struct{}{
@@ -71,7 +71,7 @@ func TestVerifyRegions(t *testing.T) {
 		},
 		{
 			name: "region_matches_all",
-			data: &database.Publish{Regions: []string{"US", "CA"}},
+			data: &model.Publish{Regions: []string{"US", "CA"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 				AllowedRegions: map[string]struct{}{
@@ -82,7 +82,7 @@ func TestVerifyRegions(t *testing.T) {
 		},
 		{
 			name: "region_matches_some",
-			data: &database.Publish{Regions: []string{"US", "MX"}},
+			data: &model.Publish{Regions: []string{"US", "MX"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 				AllowedRegions: map[string]struct{}{
@@ -94,7 +94,7 @@ func TestVerifyRegions(t *testing.T) {
 		},
 		{
 			name: "region_matches_none",
-			data: &database.Publish{Regions: []string{"MX"}},
+			data: &model.Publish{Regions: []string{"MX"}},
 			cfg: &authorizedapp.AuthorizedApp{
 				AppPackageName: appPkgName,
 				AllowedRegions: map[string]struct{}{
@@ -123,27 +123,27 @@ func TestVerifySafetyNet(t *testing.T) {
 	}
 
 	cases := []struct {
-		Data              *database.Publish
+		Data              *model.Publish
 		Msg               string
 		Cfg               *authorizedapp.AuthorizedApp
 		AttestationResult error
 	}{
 		{
 			// With no configuration, return err.
-			&database.Publish{Regions: []string{"US"}},
+			&model.Publish{Regions: []string{"US"}},
 			"cannot enforce SafetyNet",
 			nil,
 			nil,
 		}, {
 			// Verify when Validate Attestation Passes, return nil.
-			&database.Publish{Regions: []string{"US"}},
+			&model.Publish{Regions: []string{"US"}},
 			"",
 			allRegions,
 			nil,
 		}, {
 			// Verify when ValidateAttestation raises err, with safety check
 			// enabled, return err.
-			&database.Publish{Regions: []string{"US"}},
+			&model.Publish{Regions: []string{"US"}},
 			"android.ValidateAttestation: mocked",
 			allRegions,
 			fmt.Errorf("mocked"),
diff --git a/testing/enclient/enclient.go b/testing/enclient/enclient.go
index 7cc5361..b6046e4 100644
--- a/testing/enclient/enclient.go
+++ b/testing/enclient/enclient.go
@@ -9,7 +9,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/publish/model"
 )
 
 const (
@@ -70,8 +70,8 @@ func NewInterval(time int64) Interval {
 }
 
 // Creates an exposure key.
-func ExposureKey(key string, intervalNumber Interval, intervalCount int32, transmissionRisk int) database.ExposureKey {
-	return database.ExposureKey{
+func ExposureKey(key string, intervalNumber Interval, intervalCount int32, transmissionRisk int) model.ExposureKey {
+	return model.ExposureKey{
 		Key:              key,
 		IntervalNumber:   int32(intervalNumber),
 		IntervalCount:    intervalCount,