Depsync cleanup (#505)

Closes #302

This package still needs a lot of work, but I feel like this is a good stopping point.
In a follow up, I plan on writing some tests, and refactoring some of the ways the sync.Watcher goroutines are managed.

cmd/mdmctl/apply_dep_autoassigner.go

@@ -5,7 +5,7 @@ import (
 	"flag"
 	"fmt"
 
-	"github.com/micromdm/micromdm/dep/depsync"
+	"github.com/micromdm/micromdm/platform/dep/sync"
 	"github.com/pkg/errors"
 )
 
@@ -24,7 +24,7 @@ func (cmd *applyCommand) applyDEPAutoAssigner(args []string) error {
 		return errors.New("bad input: must provide both -filter and -uuid")
 	}
 
-	assigner := depsync.AutoAssigner{*flFilter, *flProfileUUID}
+	assigner := sync.AutoAssigner{*flFilter, *flProfileUUID}
 
 	err := cmd.depsyncsvc.ApplyAutoAssigner(context.TODO(), &assigner)
 	if err != nil {

cmd/mdmctl/apply_dep_profile.go

@@ -13,8 +13,8 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/micromdm/micromdm/dep"
-	"github.com/micromdm/micromdm/dep/depsync"
 	"github.com/micromdm/micromdm/pkg/crypto"
+	"github.com/micromdm/micromdm/platform/dep/sync"
 )
 
 func certificatesFromURL(serverURL string, insecure bool) ([]*x509.Certificate, error) {
@@ -101,7 +101,7 @@ func (cmd *applyCommand) applyDEPProfile(args []string) error {
 	fmt.Printf("Defined DEP Profile with UUID %s\n", resp.ProfileUUID)
 
 	if *flFilter != "" {
-		assigner := depsync.AutoAssigner{*flFilter, resp.ProfileUUID}
+		assigner := sync.AutoAssigner{*flFilter, resp.ProfileUUID}
 		err := cmd.depsyncsvc.ApplyAutoAssigner(context.TODO(), &assigner)
 		if err != nil {
 			return errors.Wrap(err, "set auto-assigner")

cmd/mdmctl/setup.go

@@ -4,11 +4,11 @@ import (
 	"github.com/go-kit/kit/log"
 	httptransport "github.com/go-kit/kit/transport/http"
 
-	"github.com/micromdm/micromdm/dep/depsync"
 	"github.com/micromdm/micromdm/platform/appstore"
 	"github.com/micromdm/micromdm/platform/blueprint"
 	"github.com/micromdm/micromdm/platform/config"
 	"github.com/micromdm/micromdm/platform/dep"
+	"github.com/micromdm/micromdm/platform/dep/sync"
 	"github.com/micromdm/micromdm/platform/device"
 	"github.com/micromdm/micromdm/platform/profile"
 	"github.com/micromdm/micromdm/platform/remove"
@@ -24,7 +24,7 @@ type remoteServices struct {
 	configsvc    config.Service
 	appsvc       appstore.Service
 	depsvc       dep.Service
-	depsyncsvc   depsync.Service
+	depsyncsvc   sync.Service
 }
 
 func setupClient(logger log.Logger) (*remoteServices, error) {
@@ -89,7 +89,7 @@ func setupClient(logger log.Logger) (*remoteServices, error) {
 		return nil, err
 	}
 
-	depsyncsvc, err := depsync.NewHTTPClient(
+	depsyncsvc, err := sync.NewHTTPClient(
 		cfg.ServerURL, cfg.APIToken, logger,
 		httptransport.SetClient(skipVerifyHTTPClient(cfg.SkipVerify)))
 	if err != nil {

cmd/micromdm/serve.go

@@ -29,7 +29,6 @@ import (
 	"github.com/pkg/errors"
 	"golang.org/x/crypto/acme/autocert"
 
-	"github.com/micromdm/micromdm/dep/depsync"
 	"github.com/micromdm/micromdm/mdm"
 	"github.com/micromdm/micromdm/mdm/enroll"
 	httputil2 "github.com/micromdm/micromdm/pkg/httputil"
@@ -41,6 +40,7 @@ import (
 	"github.com/micromdm/micromdm/platform/command"
 	"github.com/micromdm/micromdm/platform/config"
 	depapi "github.com/micromdm/micromdm/platform/dep"
+	"github.com/micromdm/micromdm/platform/dep/sync"
 	"github.com/micromdm/micromdm/platform/device"
 	devicebuiltin "github.com/micromdm/micromdm/platform/device/builtin"
 	"github.com/micromdm/micromdm/platform/profile"
@@ -253,8 +253,8 @@ func serve(args []string) error {
 		depEndpoints := depapi.MakeServerEndpoints(depsvc, basicAuthEndpointMiddleware)
 		depapi.RegisterHTTPHandlers(r, depEndpoints, options...)
 
-		depsyncEndpoints := depsync.MakeServerEndpoints(depsync.NewService(syncer), basicAuthEndpointMiddleware)
-		depsync.RegisterHTTPHandlers(r, depsyncEndpoints, options...)
+		depsyncEndpoints := sync.MakeServerEndpoints(sync.NewService(syncer, sm.SyncDB), basicAuthEndpointMiddleware)
+		sync.RegisterHTTPHandlers(r, depsyncEndpoints, options...)
 	} else {
 		mainLogger.Log("msg", "no api key specified")
 	}

dep/depsync/config.go

@@ -1,95 +0,0 @@
-package depsync
-
-import (
-	"encoding/json"
-
-	"github.com/boltdb/bolt"
-	"github.com/pkg/errors"
-)
-
-type config struct {
-	*bolt.DB
-	Cursor cursor `json:"cursor"`
-}
-
-func (cfg *config) Save() error {
-	err := cfg.Update(func(tx *bolt.Tx) error {
-		bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket))
-		if err != nil {
-			return err
-		}
-		v, err := json.Marshal(cfg)
-		if err != nil {
-			return err
-		}
-		return bkt.Put([]byte("configuration"), v)
-	})
-	return errors.Wrap(err, "saving dep sync cursor")
-}
-
-func (cfg *config) saveAutoAssigner(assigner *AutoAssigner) error {
-	if assigner.Filter != "*" {
-		return errors.New("only '*' filter auto-assigners supported")
-	}
-	err := cfg.Update(func(tx *bolt.Tx) error {
-		b, err := tx.CreateBucketIfNotExists([]byte(AutoAssignBucket))
-		if err != nil {
-			return err
-		}
-		return b.Put([]byte(assigner.Filter), []byte(assigner.ProfileUUID))
-	})
-	return errors.Wrap(err, "saving auto-assigner")
-}
-
-func (cfg *config) loadAutoAssigners() ([]*AutoAssigner, error) {
-	assigners := []*AutoAssigner{}
-	err := cfg.View(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte(AutoAssignBucket))
-		if b == nil { // bucket doesn't exist yet
-			return nil
-		}
-
-		return b.ForEach(func(k, v []byte) error {
-			assigners = append(assigners, &AutoAssigner{
-				Filter:      string(k),
-				ProfileUUID: string(v),
-			})
-			return nil
-		})
-	})
-	return assigners, errors.Wrap(err, "loading auto-assigners")
-}
-
-func (cfg *config) deleteAutoAssigner(filter string) error {
-	return cfg.Update(func(tx *bolt.Tx) error {
-		b := tx.Bucket([]byte(AutoAssignBucket))
-		if b == nil { // bucket doesn't exist yet
-			return nil
-		}
-		return b.Delete([]byte(filter))
-	})
-}
-
-func LoadConfig(db *bolt.DB) (*config, error) {
-	conf := config{DB: db}
-	err := db.Update(func(tx *bolt.Tx) error {
-		bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket))
-		if err != nil {
-			return err
-		}
-
-		v := bkt.Get([]byte("configuration"))
-		if v == nil {
-			return nil
-		}
-		if err := json.Unmarshal(v, &conf); err != nil {
-			return err
-		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return &conf, nil
-}

dep/depsync/service.go

@@ -1,16 +0,0 @@
-package depsync
-
-import (
-	"context"
-)
-
-type Service interface {
-	SyncNow(context.Context) error
-	ApplyAutoAssigner(context.Context, *AutoAssigner) error
-	GetAutoAssigners(context.Context) ([]*AutoAssigner, error)
-	RemoveAutoAssigner(context.Context, string) error
-}
-
-type DEPSyncService struct {
-	syncer Syncer
-}

go.sum

@@ -22,6 +22,7 @@ github.com/gorilla/mux v1.6.2 h1:Pgr17XVTNXAk3q/r4CpKzC5xBM/qW1uVLV+IhRZpIIk=
 github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/groob/finalizer v0.0.0-20170707115354-4c2ed49aabda h1:5ikpG9mYCMFiZX0nkxoV6aU2IpCHPdws3gCNgdZeEV0=
 github.com/groob/finalizer v0.0.0-20170707115354-4c2ed49aabda/go.mod h1:MyndkAZd5rUMdNogn35MWXBX1UiBigrU8eTj8DoAC2c=
+github.com/groob/pkcs7 v0.0.0-20180824154052-36585635cb64 h1:1ALD84dEnUxPKZENhUAeQ0tuJ+s3PuL85pV95B0Ekfk=
 github.com/groob/pkcs7 v0.0.0-20180824154052-36585635cb64/go.mod h1:mEOMQ8C7oeXY3LnE2jy4UkLAqrW9rrpwiP5U4hVV+MY=
 github.com/groob/plist v0.0.0-20180203051248-dd56909aee38 h1:afbUddvIjPRC7XHHgeSTRfzZtIxEsSl4VCxumLBGDJU=
 github.com/groob/plist v0.0.0-20180203051248-dd56909aee38/go.mod h1:qg2Nek0ND/hIr+nY8H1oVqEW2cLzVVNaAQ0QexOyjyc=

platform/dep/sync/apply_autoassigner.go

@@ -1,17 +1,18 @@
-package depsync
+package sync
 
 import (
 	"context"
 	"net/http"
 
 	"github.com/go-kit/kit/endpoint"
+	"github.com/pkg/errors"
 
 	"github.com/micromdm/micromdm/pkg/httputil"
 )
 
 func (s DEPSyncService) ApplyAutoAssigner(ctx context.Context, aa *AutoAssigner) error {
-	conf := s.syncer.GetConfig()
-	return conf.saveAutoAssigner(aa)
+	err := s.db.SaveAutoAssigner(aa)
+	return errors.Wrap(err, "saving AutoAssigner")
 }
 
 type applyAutoAssignerRequest struct {

platform/dep/sync/builtin/db.go

@@ -0,0 +1,116 @@
+package builtin
+
+import (
+	"encoding/json"
+
+	"github.com/boltdb/bolt"
+	"github.com/pkg/errors"
+
+	"github.com/micromdm/micromdm/platform/dep/sync"
+)
+
+const (
+	ConfigBucket     = "mdm.DEPConfig"
+	AutoAssignBucket = "mdm.DEPAutoAssign"
+)
+
+type DB struct {
+	*bolt.DB
+}
+
+func NewDB(db *bolt.DB) (*DB, error) {
+	err := db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket))
+		if err != nil {
+			return err
+		}
+		_, err = tx.CreateBucketIfNotExists([]byte(AutoAssignBucket))
+		return err
+	})
+	if err != nil {
+		return nil, errors.Wrapf(err, "creating %s bucket", ConfigBucket)
+	}
+	datastore := &DB{DB: db}
+	return datastore, nil
+}
+
+func (db *DB) LoadCursor() (*sync.Cursor, error) {
+	var cursor = struct {
+		Cursor sync.Cursor `json:"cursor"`
+	}{}
+	err := db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket([]byte(ConfigBucket))
+		v := bkt.Get([]byte("configuration"))
+		if v == nil {
+			return nil // TODO add notfound
+		}
+		err := json.Unmarshal(v, &cursor)
+		return errors.Wrap(err, "unmarshal dep cursor")
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "load cursor from bolt")
+	}
+	return &cursor.Cursor, nil
+}
+
+func (db *DB) SaveCursor(c sync.Cursor) error {
+	err := db.Update(func(tx *bolt.Tx) error {
+		bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket))
+		if err != nil {
+			return err
+		}
+		// use anonymous struct to preserve old structure.
+		var cursor = struct {
+			Cursor sync.Cursor `json:"cursor"`
+		}{Cursor: c}
+		v, err := json.Marshal(&cursor)
+		if err != nil {
+			return err
+		}
+		return bkt.Put([]byte("configuration"), v)
+	})
+	return errors.Wrap(err, "saving dep sync cursor")
+}
+
+func (db *DB) SaveAutoAssigner(a *sync.AutoAssigner) error {
+	if a.Filter != "*" {
+		return errors.New("only '*' filter auto-assigners supported")
+	}
+	err := db.Update(func(tx *bolt.Tx) error {
+		b, err := tx.CreateBucketIfNotExists([]byte(AutoAssignBucket))
+		if err != nil {
+			return err
+		}
+		return b.Put([]byte(a.Filter), []byte(a.ProfileUUID))
+	})
+	return errors.Wrap(err, "saving auto-assigner")
+}
+
+func (db *DB) DeleteAutoAssigner(filter string) error {
+	return db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte(AutoAssignBucket))
+		if b == nil { // bucket doesn't exist yet
+			return nil
+		}
+		return b.Delete([]byte(filter))
+	})
+}
+
+func (db *DB) LoadAutoAssigners() ([]sync.AutoAssigner, error) {
+	var aa []sync.AutoAssigner
+	err := db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte(AutoAssignBucket))
+		if b == nil { // bucket doesn't exist yet
+			return nil
+		}
+
+		return b.ForEach(func(k, v []byte) error {
+			aa = append(aa, sync.AutoAssigner{
+				Filter:      string(k),
+				ProfileUUID: string(v),
+			})
+			return nil
+		})
+	})
+	return aa, errors.Wrap(err, "loading auto-assigners")
+}

platform/dep/sync/client.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"net/url"

platform/dep/sync/depsync.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"context"
@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/boltdb/bolt"
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
@@ -18,47 +17,81 @@ import (
 )
 
 const (
-	SyncTopic        = "mdm.DepSync"
-	ConfigBucket     = "mdm.DEPConfig"
-	AutoAssignBucket = "mdm.DEPAutoAssign"
+	SyncTopic = "mdm.DepSync"
 
 	syncDuration        = 30 * time.Minute
 	cursorValidDuration = 7 * 24 * time.Hour
 )
 
-type Syncer interface {
-	SyncNow()
-	GetConfig() *config // TODO: #302
+type Syncer interface{ SyncNow() }
+
+type WatcherDB interface {
+	LoadCursor() (*Cursor, error)
+	SaveCursor(c Cursor) error
+	LoadAutoAssigners() ([]AutoAssigner, error)
 }
 
-type AutoAssigner struct {
-	Filter      string `json:"filter"`
-	ProfileUUID string `json:"profile_uuid"`
-}
-
-type watcher struct {
+type Watcher struct {
 	mtx    sync.RWMutex
 	logger log.Logger
 	client Client
 
 	publisher pubsub.Publisher
-	conf      *config
+	db        WatcherDB
 	startSync chan bool
 	syncNow   chan bool
+
+	cursor Cursor
 }
 
-type cursor struct {
-	Value     string    `json:"value"`
-	CreatedAt time.Time `json:"created_at"`
-}
-
-// A cursor is valid for a week.
-func (c cursor) Valid() bool {
-	expiration := time.Now().Add(cursorValidDuration)
-	if c.CreatedAt.After(expiration) {
-		return false
+func NewWatcher(db WatcherDB, pub pubsub.PublishSubscriber, opts ...Option) (*Watcher, error) {
+	w := Watcher{
+		logger:    log.NewNopLogger(),
+		db:        db,
+		publisher: pub,
+		startSync: make(chan bool),
+		syncNow:   make(chan bool),
 	}
-	return true
+	for _, optFn := range opts {
+		optFn(&w)
+	}
+
+	cursor, err := w.db.LoadCursor()
+	if err != nil {
+		return nil, err
+	}
+	if cursor.Valid() {
+		level.Debug(w.logger).Log("msg", "loaded DEP config", "cursor", cursor.Value)
+		w.cursor = *cursor
+	}
+
+	if err := w.updateClient(pub); err != nil {
+		return nil, err
+	}
+
+	saveCursor := func() {
+		if err := db.SaveCursor(w.cursor); err != nil {
+			level.Info(w.logger).Log("err", err, "msg", "saving cursor")
+			return
+		}
+		level.Info(w.logger).Log("msg", "saved DEP config", "cursor", w.cursor.Value)
+	}
+
+	go func() {
+		defer saveCursor()
+		if w.client == nil {
+			panic("boom")
+			// block until we have a DEP client to start sync process
+			level.Info(w.logger).Log("msg", "waiting for DEP token to be added before starting sync")
+			<-w.startSync
+		}
+		err := w.Run()
+		// the DEP sync should never end without an error, but log
+		// unconditionally anyway so we never silently stop watching
+		level.Info(w.logger).Log("err", err, "msg", "DEP watcher stopped")
+	}()
+
+	return &w, nil
 }
 
 type Client interface {
@@ -67,71 +100,21 @@ type Client interface {
 	AssignProfile(string, ...string) (*dep.ProfileResponse, error)
 }
 
-type Option func(*watcher)
+type Option func(*Watcher)
 
 func WithClient(client Client) Option {
-	return func(w *watcher) {
+	return func(w *Watcher) {
 		w.client = client
 	}
 }
 
 func WithLogger(logger log.Logger) Option {
-	return func(w *watcher) {
+	return func(w *Watcher) {
 		w.logger = logger
 	}
 }
 
-func New(pub pubsub.PublishSubscriber, db *bolt.DB, logger log.Logger, opts ...Option) (Syncer, error) {
-	conf, err := LoadConfig(db)
-	if err != nil {
-		return nil, err
-	}
-	if conf.Cursor.Valid() {
-		level.Info(logger).Log("msg", "loaded DEP config", "cursor", conf.Cursor.Value)
-	} else {
-		conf.Cursor.Value = ""
-	}
-
-	sync := &watcher{
-		logger:    log.NewNopLogger(),
-		publisher: pub,
-		conf:      conf,
-		startSync: make(chan bool),
-		syncNow:   make(chan bool),
-	}
-
-	for _, opt := range opts {
-		opt(sync)
-	}
-
-	if err := sync.updateClient(pub); err != nil {
-		return nil, err
-	}
-
-	saveCursor := func() {
-		if err := conf.Save(); err != nil {
-			level.Info(logger).Log("err", err, "msg", "saving cursor")
-			return
-		}
-		level.Info(logger).Log("msg", "saved DEP config", "cursor", conf.Cursor.Value)
-	}
-
-	go func() {
-		defer saveCursor()
-		if sync.client == nil {
-			// block until we have a DEP client to start sync process
-			level.Info(logger).Log("msg", "waiting for DEP token to be added before starting sync")
-			<-sync.startSync
-		}
-		err := sync.Run()
-		// the DEP sync should never end without an error, but log
-		// unconditionally anyway so we never silently stop watching
-		level.Info(logger).Log("err", err, "msg", "DEP watcher stopped")
-	}()
-	return sync, nil
-}
-
-func (w *watcher) updateClient(pubsub pubsub.Subscriber) error {
+func (w *Watcher) updateClient(pubsub pubsub.Subscriber) error {
 	tokenAdded, err := pubsub.Subscribe(context.TODO(), "token-events", conf.DEPTokenTopic)
 	if err != nil {
 		return err
@@ -163,7 +146,7 @@ func (w *watcher) updateClient(pubsub pubsub.Subscriber) error {
 	return nil
 }
 
-func (w *watcher) SyncNow() {
+func (w *Watcher) SyncNow() {
 	if w.client == nil {
 		level.Info(w.logger).Log("msg", "waiting for DEP token to be added before starting sync")
 		return
@@ -171,10 +154,6 @@ func (w *watcher) SyncNow() {
 	w.syncNow <- true
 }
 
-func (w *watcher) GetConfig() *config {
-	return w.conf
-}
-
 // TODO this needs to be a proper error in the micromdm/dep package.
 func isCursorExhausted(err error) bool {
 	return strings.Contains(err.Error(), "EXHAUSTED_CURSOR")
@@ -190,12 +169,12 @@ func isCursorInvalid(err error) bool {
 
 // Process DEP messages and pull out filter-matching serial numbers
 // associated to profile UUIDs for auto-assignment.
-func (w *watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]string, error) {
+func (w *Watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]string, error) {
 	// load auto-assigners every run to make sure we get the latest set of
 	// auto-assigner profile UUIDs/filters. Note this makes every *watcher
 	// (i.e. every DEP sync instance) share the current DB set of auto-
 	// assigners. perhaps to refactor to be more separated.
-	assigners, err := w.conf.loadAutoAssigners()
+	assigners, err := w.db.LoadAutoAssigners()
 	if err != nil {
 		return nil, err
 	}
@@ -225,7 +204,7 @@ func (w *watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]st
 	return assigned, nil
 }
 
-func (w *watcher) processAutoAssign(devices []dep.Device) error {
+func (w *Watcher) processAutoAssign(devices []dep.Device) error {
 	assignments, err := w.filteredAutoAssignments(devices)
 	if err != nil {
 		return err
@@ -267,7 +246,7 @@ func (w *watcher) processAutoAssign(devices []dep.Device) error {
 	return nil
 }
 
-func (w *watcher) publishAndProcessDevices(devices []dep.Device) error {
+func (w *Watcher) publishAndProcessDevices(devices []dep.Device) error {
 	e := NewEvent(devices)
 	data, err := MarshalEvent(e)
 	if err != nil {
@@ -291,21 +270,21 @@ func (w *watcher) publishAndProcessDevices(devices []dep.Device) error {
 	return nil
 }
 
-func (w *watcher) Run() error {
+func (w *Watcher) Run() error {
 	ticker := time.NewTicker(syncDuration).C
 FETCH:
 	for {
-		resp, err := w.client.FetchDevices(dep.Limit(100), dep.Cursor(w.conf.Cursor.Value))
+		resp, err := w.client.FetchDevices(dep.Limit(100), dep.Cursor(w.cursor.Value))
 		if err != nil && isCursorExhausted(err) {
 			goto SYNC
 		} else if err != nil && isCursorInvalid(err) {
 			level.Info(w.logger).Log(
 				"msg", "DEP fetch cursor response",
-				"cursor", w.conf.Cursor.Value,
+				"cursor", w.cursor.Value,
 				"err", err,
 				"msg", "retrying DEP fetch with empty cursor",
 			)
-			w.conf.Cursor.Value = ""
+			w.cursor.Value = ""
 			goto FETCH
 		} else if err != nil {
 			return err
@@ -317,8 +296,8 @@ FETCH:
 			"fetched", resp.FetchedUntil,
 			"devices", len(resp.Devices),
 		)
-		w.conf.Cursor = cursor{Value: resp.Cursor, CreatedAt: time.Now()}
-		if err := w.conf.Save(); err != nil {
+		w.cursor = Cursor{Value: resp.Cursor, CreatedAt: time.Now()}
+		if err := w.db.SaveCursor(w.cursor); err != nil {
 			return errors.Wrap(err, "saving cursor from fetch")
 		}
 		if err := w.publishAndProcessDevices(resp.Devices); err != nil {
@@ -331,15 +310,15 @@ FETCH:
 
 SYNC:
 	for {
-		resp, err := w.client.SyncDevices(w.conf.Cursor.Value, dep.Cursor(w.conf.Cursor.Value))
+		resp, err := w.client.SyncDevices(w.cursor.Value, dep.Cursor(w.cursor.Value))
 		if err != nil && (isCursorExpired(err) || isCursorInvalid(err)) {
 			level.Info(w.logger).Log(
 				"msg", "DEP sync cursor response",
-				"cursor", w.conf.Cursor.Value,
+				"cursor", w.cursor.Value,
 				"err", err,
 				"msg", "retrying DEP fetch with empty cursor",
 			)
-			w.conf.Cursor.Value = ""
+			w.cursor.Value = ""
 			goto FETCH
 		} else if err != nil {
 			return err
@@ -351,8 +330,8 @@ SYNC:
 			"fetched", resp.FetchedUntil,
 			"devices", len(resp.Devices),
 		)
-		w.conf.Cursor = cursor{Value: resp.Cursor, CreatedAt: time.Now()}
-		if err := w.conf.Save(); err != nil {
+		w.cursor = Cursor{Value: resp.Cursor, CreatedAt: time.Now()}
+		if err := w.db.SaveCursor(w.cursor); err != nil {
 			return errors.Wrap(err, "saving cursor from sync")
 		}
 		if err := w.publishAndProcessDevices(resp.Devices); err != nil {

platform/dep/sync/event.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"time"
@@ -7,7 +7,7 @@ import (
 	uuid "github.com/satori/go.uuid"
 
 	"github.com/micromdm/micromdm/dep"
-	"github.com/micromdm/micromdm/dep/depsync/internal/depsyncproto"
+	"github.com/micromdm/micromdm/platform/dep/sync/internal/depsyncproto"
 )
 
 type Event struct {

platform/dep/sync/get_autoassigners.go

@@ -1,22 +1,23 @@
-package depsync
+package sync
 
 import (
 	"context"
 	"net/http"
 
 	"github.com/go-kit/kit/endpoint"
+	"github.com/pkg/errors"
 
 	"github.com/micromdm/micromdm/pkg/httputil"
 )
 
-func (s DEPSyncService) GetAutoAssigners(ctx context.Context) ([]*AutoAssigner, error) {
-	conf := s.syncer.GetConfig()
-	return conf.loadAutoAssigners()
+func (s DEPSyncService) GetAutoAssigners(ctx context.Context) ([]AutoAssigner, error) {
+	aa, err := s.db.LoadAutoAssigners()
+	return aa, errors.Wrap(err, "get AutoAssigners")
 }
 
 type getAutoAssignersResponse struct {
-	AutoAssigners []*AutoAssigner `json:"autoassigners"`
-	Err           error           `json:"err,omitempty"`
+	AutoAssigners []AutoAssigner `json:"autoassigners"`
+	Err           error          `json:"err,omitempty"`
 }
 
 func (r getAutoAssignersResponse) Failed() error { return r.Err }
@@ -37,7 +38,7 @@ func decodeGetAutoAssignersResponse(ctx context.Context, r *http.Response) (inte
 	return req, err
 }
 
-func (e Endpoints) GetAutoAssigners(ctx context.Context) ([]*AutoAssigner, error) {
+func (e Endpoints) GetAutoAssigners(ctx context.Context) ([]AutoAssigner, error) {
 	resp, err := e.GetAutoAssignersEndpoint(ctx, nil)
 	if err != nil {
 		return nil, err

platform/dep/sync/remove_autoassigners.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"context"
@@ -6,11 +6,12 @@ import (
 
 	"github.com/go-kit/kit/endpoint"
 	"github.com/micromdm/micromdm/pkg/httputil"
+	"github.com/pkg/errors"
 )
 
 func (s DEPSyncService) RemoveAutoAssigner(ctx context.Context, filter string) error {
-	conf := s.syncer.GetConfig()
-	return conf.deleteAutoAssigner(filter)
+	err := s.db.DeleteAutoAssigner(filter)
+	return errors.Wrap(err, "remove AutoAssigner")
 }
 
 type removeAutoAssignerRequest struct {

platform/dep/sync/server.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"github.com/go-kit/kit/endpoint"
@@ -8,8 +8,8 @@ import (
 	"github.com/micromdm/micromdm/pkg/httputil"
 )
 
-func NewService(syncer Syncer) *DEPSyncService {
-	return &DEPSyncService{syncer: syncer}
+func NewService(syncer Syncer, db DB) *DEPSyncService {
+	return &DEPSyncService{syncer: syncer, db: db}
 }
 
 type Endpoints struct {

platform/dep/sync/service.go

@@ -0,0 +1,40 @@
+package sync
+
+import (
+	"context"
+	"time"
+)
+
+type Service interface {
+	SyncNow(context.Context) error
+	ApplyAutoAssigner(context.Context, *AutoAssigner) error
+	GetAutoAssigners(context.Context) ([]AutoAssigner, error)
+	RemoveAutoAssigner(context.Context, string) error
+}
+
+type DB interface {
+	SaveAutoAssigner(a *AutoAssigner) error
+	LoadAutoAssigners() ([]AutoAssigner, error)
+	DeleteAutoAssigner(filter string) error
+}
+
+type DEPSyncService struct {
+	db     DB
+	syncer Syncer
+}
+
+type Cursor struct {
+	Value     string    `json:"value"`
+	CreatedAt time.Time `json:"created_at"`
+}
+
+// A cursor is valid for a week.
+func (c Cursor) Valid() bool {
+	expiration := time.Now().Add(cursorValidDuration)
+	return c.CreatedAt.Before(expiration)
+}
+
+type AutoAssigner struct {
+	Filter      string `json:"filter"`
+	ProfileUUID string `json:"profile_uuid"`
+}

platform/dep/sync/syncnow.go

@@ -1,4 +1,4 @@
-package depsync
+package sync
 
 import (
 	"context"

platform/device/worker.go

@@ -9,8 +9,8 @@ import (
 	"github.com/pkg/errors"
 	uuid "github.com/satori/go.uuid"
 
-	"github.com/micromdm/micromdm/dep/depsync"
 	"github.com/micromdm/micromdm/mdm"
+	"github.com/micromdm/micromdm/platform/dep/sync"
 	"github.com/micromdm/micromdm/platform/pubsub"
 )
 
@@ -48,9 +48,9 @@ func (w *Worker) Run(ctx context.Context) error {
 	if err != nil {
 		return errors.Wrapf(err, "subscribing %s to %s", subscription, mdm.CheckoutTopic)
 	}
-	depSyncEvents, err := w.ps.Subscribe(ctx, subscription, depsync.SyncTopic)
+	depSyncEvents, err := w.ps.Subscribe(ctx, subscription, sync.SyncTopic)
 	if err != nil {
-		return errors.Wrapf(err, "subscribing %s to %s", subscription, depsync.SyncTopic)
+		return errors.Wrapf(err, "subscribing %s to %s", subscription, sync.SyncTopic)
 	}
 	connectEvents, err := w.ps.Subscribe(ctx, subscription, mdm.ConnectTopic)
 	if err != nil {
@@ -84,8 +84,8 @@ func (w *Worker) Run(ctx context.Context) error {
 }
 
 func (w *Worker) updateFromDEPSync(ctx context.Context, message []byte) error {
-	var ev depsync.Event
-	if err := depsync.UnmarshalEvent(message, &ev); err != nil {
+	var ev sync.Event
+	if err := sync.UnmarshalEvent(message, &ev); err != nil {
 		return errors.Wrap(err, "unmarshal depsync event")
 	}
 	level.Debug(w.logger).Log(

server/server.go

@@ -23,7 +23,6 @@ import (
 	"golang.org/x/crypto/pkcs12"
 
 	"github.com/micromdm/micromdm/dep"
-	"github.com/micromdm/micromdm/dep/depsync"
 	"github.com/micromdm/micromdm/mdm"
 	"github.com/micromdm/micromdm/mdm/enroll"
 	"github.com/micromdm/micromdm/pkg/crypto"
@@ -32,6 +31,8 @@ import (
 	"github.com/micromdm/micromdm/platform/command"
 	"github.com/micromdm/micromdm/platform/config"
 	configbuiltin "github.com/micromdm/micromdm/platform/config/builtin"
+	"github.com/micromdm/micromdm/platform/dep/sync"
+	syncbuiltin "github.com/micromdm/micromdm/platform/dep/sync/builtin"
 	"github.com/micromdm/micromdm/platform/profile"
 	profilebuiltin "github.com/micromdm/micromdm/platform/profile/builtin"
 	"github.com/micromdm/micromdm/platform/pubsub"
@@ -60,6 +61,7 @@ type Server struct {
 	RemoveDB            block.Store
 	CommandWebhookURL   string
 	DEPClient           *dep.Client
+	SyncDB              *syncbuiltin.DB
 
 	PushService     *push.Service // bufford push
 	APNSPushService apns.Service
@@ -398,17 +400,23 @@ func (c *Server) setupDepClient() error {
 	return nil
 }
 
-func (c *Server) CreateDEPSyncer(logger log.Logger) (depsync.Syncer, error) {
+func (c *Server) CreateDEPSyncer(logger log.Logger) (sync.Syncer, error) {
 	client := c.DEPClient
-	opts := []depsync.Option{
-		depsync.WithLogger(log.With(logger, "component", "depsync")),
+	opts := []sync.Option{
+		sync.WithLogger(log.With(logger, "component", "depsync")),
 	}
 	if client != nil {
-		opts = append(opts, depsync.WithClient(client))
+		opts = append(opts, sync.WithClient(client))
 	}
 
-	var syncer depsync.Syncer
-	syncer, err := depsync.New(c.PubClient, c.DB, logger, opts...)
+	syncdb, err := syncbuiltin.NewDB(c.DB)
+	if err != nil {
+		return nil, err
+	}
+	c.SyncDB = syncdb
+
+	var syncer sync.Syncer
+	syncer, err = sync.NewWatcher(c.SyncDB, c.PubClient, opts...)
 	if err != nil {
 		return nil, err
 	}