diff --git a/api.go b/api.go
index 73399d7..1fceebb 100644
--- a/api.go
+++ b/api.go
@@ -24,32 +24,45 @@
 
 const tokenEndpoint = "/accesstoken"
 
-func InitAPI(services apid.Services) {
-	services.API().HandleFunc(tokenEndpoint, getAccessToken).Methods("GET")
+const (
+	// long-polling timeout from http header
+	parBlock = "block"
+	// long-polling tag used for comparision
+	// if tag fails to match, new token is returned immediately
+	parTag = "If-None-Match"
+)
+
+type ApiManager struct {
+	tokenMan tokenManager
+	endpoint string
 }
 
-func getAccessToken(w http.ResponseWriter, r *http.Request) {
-	b := r.URL.Query().Get("block")
+func (a *ApiManager) InitAPI(api apid.APIService) {
+	api.HandleFunc(a.endpoint, a.getAccessToken).Methods("GET")
+}
+
+func (a *ApiManager) getAccessToken(w http.ResponseWriter, r *http.Request) {
+	b := r.URL.Query().Get(parBlock)
 	var timeout int
 	if b != "" {
 		var err error
 		timeout, err = strconv.Atoi(b)
-		if err != nil {
+		if err != nil || timeout < 0 {
 			writeError(w, http.StatusBadRequest, "bad block value, must be number of seconds")
 			return
 		}
 	}
 	log.Debugf("api timeout: %d", timeout)
-	ifNoneMatch := r.Header.Get("If-None-Match")
-
-	if apidTokenManager.getBearerToken() != ifNoneMatch {
-		w.Write([]byte(apidTokenManager.getBearerToken()))
+	ifNoneMatch := r.Header.Get(parTag)
+	log.Debugf("ifNoneMatch: %s", ifNoneMatch)
+	if a.tokenMan.getBearerToken() != ifNoneMatch {
+		w.Write([]byte(a.tokenMan.getBearerToken()))
 		return
 	}
 
 	select {
-	case <-apidTokenManager.getTokenReadyChannel():
-		w.Write([]byte(apidTokenManager.getBearerToken()))
+	case <-a.tokenMan.getTokenReadyChannel():
+		w.Write([]byte(a.tokenMan.getBearerToken()))
 	case <-time.After(time.Duration(timeout) * time.Second):
 		w.WriteHeader(http.StatusNotModified)
 	}
diff --git a/api_test.go b/api_test.go
new file mode 100644
index 0000000..d46ac4c
--- /dev/null
+++ b/api_test.go
@@ -0,0 +1,125 @@
+// Copyright 2017 Google Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apidApigeeSync
+
+import (
+	"fmt"
+	"github.com/apid/apid-core"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+)
+
+const (
+	apiTestUrl = "http://127.0.0.1:9000"
+)
+
+var _ = Describe("API Manager", func() {
+	testCount := 0
+	var testApiMan *ApiManager
+	var dummyTokenMan *dummyTokenManager
+	var client *http.Client
+	BeforeEach(func() {
+		testCount++
+		dummyTokenMan = &dummyTokenManager{
+			token:          fmt.Sprintf("test_token_%d", testCount),
+			tokenReadyChan: make(chan bool, 1),
+		}
+		testApiMan = &ApiManager{
+			endpoint: tokenEndpoint + strconv.Itoa(testCount),
+			tokenMan: dummyTokenMan,
+		}
+		testApiMan.InitAPI(apid.API())
+		time.Sleep(100 * time.Millisecond)
+		client = &http.Client{}
+	})
+
+	clientGet := func(path string, pars map[string][]string, header map[string][]string) (int, []byte) {
+		uri, err := url.Parse(apiTestUrl + path)
+		Expect(err).Should(Succeed())
+		query := url.Values(pars)
+		uri.RawQuery = query.Encode()
+		httpReq, err := http.NewRequest("GET", uri.String(), nil)
+		httpReq.Header = http.Header(header)
+		Expect(err).Should(Succeed())
+		res, err := client.Do(httpReq)
+		Expect(err).Should(Succeed())
+		defer res.Body.Close()
+		responseBody, err := ioutil.ReadAll(res.Body)
+		Expect(err).Should(Succeed())
+		return res.StatusCode, responseBody
+	}
+
+	It("should get token without long-polling", func() {
+		code, res := clientGet(testApiMan.endpoint, nil, nil)
+		Expect(code).Should(Equal(http.StatusOK))
+		Expect(string(res)).Should(Equal(dummyTokenMan.token))
+	})
+
+	It("should get bad request for invalid timeout", func() {
+		code, _ := clientGet(testApiMan.endpoint, map[string][]string{
+			parBlock: {"invalid"},
+		}, map[string][]string{
+			parTag: {dummyTokenMan.getBearerToken()},
+		})
+		Expect(code).Should(Equal(http.StatusBadRequest))
+
+		code, _ = clientGet(testApiMan.endpoint, map[string][]string{
+			parBlock: {"-1"},
+		}, map[string][]string{
+			parTag: {dummyTokenMan.getBearerToken()},
+		})
+		Expect(code).Should(Equal(http.StatusBadRequest))
+	})
+
+	It("should get token immediately if mismatch", func() {
+		code, res := clientGet(testApiMan.endpoint, map[string][]string{
+			parBlock: {"10"},
+		}, map[string][]string{
+			parTag: {"mismatch"},
+		})
+		Expect(code).Should(Equal(http.StatusOK))
+		Expect(string(res)).Should(Equal(dummyTokenMan.token))
+	}, 3)
+
+	It("should get StatusNotModified if timeout", func() {
+		code, _ := clientGet(testApiMan.endpoint, map[string][]string{
+			parBlock: {"1"},
+		}, map[string][]string{
+			parTag: {dummyTokenMan.getBearerToken()},
+		})
+		Expect(code).Should(Equal(http.StatusNotModified))
+	}, 3)
+
+	It("should do long-polling", func() {
+		go func() {
+			time.Sleep(1)
+			dummyTokenMan.token = "new_token"
+			dummyTokenMan.tokenReadyChan <- true
+		}()
+		code, res := clientGet(testApiMan.endpoint, map[string][]string{
+			parBlock: {"10"},
+		}, map[string][]string{
+			parTag: {dummyTokenMan.getBearerToken()},
+		})
+		Expect(code).Should(Equal(http.StatusOK))
+		Expect(string(res)).Should(Equal(dummyTokenMan.getBearerToken()))
+	}, 3)
+
+})
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go
index 5055f68..47c5498 100644
--- a/apigeeSync_suite_test.go
+++ b/apigeeSync_suite_test.go
@@ -19,72 +19,70 @@
 	. "github.com/onsi/gomega"
 
 	"io/ioutil"
-	"net/http/httptest"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/apid/apid-core"
-
+	"github.com/apid/apid-core/events"
 	"github.com/apid/apid-core/factory"
 )
 
-var (
-	tmpDir         string
-	testServer     *httptest.Server
-	testRouter     apid.Router
-	testMock       *MockServer
-	wipeDBAferTest bool
-)
-
 const dummyConfigValue string = "placeholder"
 const expectedClusterId = "bootstrap"
 
+var tmpDir string
+
 var _ = BeforeSuite(func() {
-	wipeDBAferTest = true
-})
-
-var _ = BeforeEach(func() {
 	apid.Initialize(factory.DefaultServicesFactory())
-
-	config = apid.Config()
 	dataService = apid.Data()
-	events = apid.Events()
-
+	config = apid.Config()
+	apiService = apid.API()
+	go apiService.Listen()
+	//dataService = apid.Data()
+	log = apid.Log().ForModule("apigeeSync")
 	var err error
-	tmpDir, err = ioutil.TempDir("", "api_test")
+	tmpDir, err = ioutil.TempDir("", "apid_test")
 	Expect(err).NotTo(HaveOccurred())
-	config.Set("local_storage_path", tmpDir)
-
+	config.Set(configLocalStoragePath, tmpDir)
 	config.Set(configProxyServerBaseURI, dummyConfigValue)
 	config.Set(configSnapServerBaseURI, dummyConfigValue)
 	config.Set(configChangeServerBaseURI, dummyConfigValue)
 	config.Set(configSnapshotProtocol, "sqlite")
 	config.Set(configPollInterval, 10*time.Millisecond)
 	config.Set(configDiagnosticMode, false)
-	config.Set(configName, "testhost")
-	config.Set(configApidClusterId, expectedClusterId)
 	config.Set(configConsumerKey, "XXXXXXX")
 	config.Set(configConsumerSecret, "YYYYYYY")
-
-	block = "0"
-	log = apid.Log().ForModule("apigeeSync")
+	config.Set(configApidInstanceID, "YYYYYYY")
 }, 3)
 
+var _ = BeforeEach(func() {
+	eventService = events.CreateService()
+	config.Set(configName, "testhost")
+	config.Set(configApidClusterId, expectedClusterId)
+	apidInfo.ClusterID = expectedClusterId
+	apidInfo.InstanceID = "YYYYYYY"
+	apidInfo.LastSnapshot = ""
+	apidInfo.IsNewInstance = true
+})
+
 var _ = AfterEach(func() {
-	apid.Events().Close()
-	lastSequence = ""
+	cleanCommonDb()
+	eventService.Close()
 })
 
 var _ = AfterSuite(func() {
-	apid.Events().Close()
-	if testServer != nil {
-		testServer.Close()
-	}
-	os.RemoveAll(tmpDir)
+	Expect(os.RemoveAll(tmpDir)).Should(Succeed())
 })
 
 func TestApigeeSync(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "ApigeeSync Suite")
 }
+
+func cleanCommonDb() {
+	db, err := dataService.DB()
+	Expect(err).Should(Succeed())
+	_, err = db.Exec(`DROP TABLE IF EXISTS APID;`)
+	Expect(err).Should(Succeed())
+}
diff --git a/apigee_sync.go b/apigee_sync.go
deleted file mode 100644
index 8a23079..0000000
--- a/apigee_sync.go
+++ /dev/null
@@ -1,140 +0,0 @@
-// Copyright 2017 Google Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package apidApigeeSync
-
-import (
-	"github.com/apid/apid-core"
-	"net/http"
-	"time"
-)
-
-const (
-	httpTimeout         = time.Minute
-	pluginTimeout       = time.Minute
-	maxIdleConnsPerHost = 10
-)
-
-var knownTables = make(map[string]bool)
-
-/*
- *  Start from existing snapshot if possible
- *  If an existing snapshot does not exist, use the apid scope to fetch
- *  all data scopes, then get a snapshot for those data scopes
- *
- *  Then, poll for changes
- */
-func bootstrap() {
-	if isOfflineMode && apidInfo.LastSnapshot == "" {
-		log.Panic("Diagnostic mode requires existent snapshot info in default DB.")
-	}
-
-	if apidInfo.LastSnapshot != "" {
-		snapshot := apidSnapshotManager.startOnLocalSnapshot(apidInfo.LastSnapshot)
-		events.EmitWithCallback(ApigeeSyncEventSelector, snapshot, func(event apid.Event) {
-			apidChangeManager.pollChangeWithBackoff()
-		})
-
-		log.Infof("Started on local snapshot: %s", snapshot.SnapshotInfo)
-		return
-	}
-
-	apidSnapshotManager.downloadBootSnapshot()
-	apidSnapshotManager.downloadDataSnapshot()
-
-	apidChangeManager.pollChangeWithBackoff()
-
-}
-
-/*
- * Call toExecute repeatedly until it does not return an error, with an exponential backoff policy
- * for retrying on errors
- */
-func pollWithBackoff(quit chan bool, toExecute func(chan bool) error, handleError func(error)) {
-
-	backoff := NewExponentialBackoff(200*time.Millisecond, config.GetDuration(configPollInterval), 2, true)
-
-	//inintialize the retry channel to start first attempt immediately
-	retry := time.After(0 * time.Millisecond)
-
-	for {
-		select {
-		case <-quit:
-			log.Info("Quit signal recieved.  Returning")
-			return
-		case <-retry:
-			start := time.Now()
-
-			err := toExecute(quit)
-			if err == nil {
-				return
-			}
-
-			if _, ok := err.(quitSignalError); ok {
-				return
-			}
-
-			end := time.Now()
-			//error encountered, since we would have returned above otherwise
-			handleError(err)
-
-			/* TODO keep this around? Imagine an immediately erroring service,
-			 *  causing many sequential requests which could pollute logs
-			 */
-			//only backoff if the request took less than one second
-			if end.After(start.Add(time.Second)) {
-				backoff.Reset()
-				retry = time.After(0 * time.Millisecond)
-			} else {
-				retry = time.After(backoff.Duration())
-			}
-		}
-	}
-}
-
-func addHeaders(req *http.Request) {
-	req.Header.Set("Authorization", "Bearer "+apidTokenManager.getBearerToken())
-	req.Header.Set("apid_instance_id", apidInfo.InstanceID)
-	req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
-	req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
-}
-
-type changeServerError struct {
-	Code string `json:"code"`
-}
-
-type quitSignalError struct {
-}
-
-type expected200Error struct {
-}
-
-type authFailError struct {
-}
-
-func (an expected200Error) Error() string {
-	return "Did not recieve OK response"
-}
-
-func (a quitSignalError) Error() string {
-	return "Signal to quit encountered"
-}
-
-func (a changeServerError) Error() string {
-	return a.Code
-}
-
-func (a authFailError) Error() string {
-	return "Authorization failed"
-}
diff --git a/apigee_sync_test.go b/apigee_sync_test.go
deleted file mode 100644
index f7144a1..0000000
--- a/apigee_sync_test.go
+++ /dev/null
@@ -1,452 +0,0 @@
-// Copyright 2017 Google Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package apidApigeeSync
-
-import (
-	"github.com/apid/apid-core"
-	"github.com/apid/apid-core/util"
-	"github.com/apigee-labs/transicator/common"
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-	"net/http"
-	"net/http/httptest"
-)
-
-var _ = Describe("Sync", func() {
-	Context("offline mode", func() {
-		var (
-			testInstanceID   = util.GenerateUUID()
-			testInstanceName = "offline-instance-name"
-			testClusterID    = "offline-cluster-id"
-			testLastSnapshot = "offline-last-snapshot"
-			testChangeMan    *dummyChangeManager
-		)
-
-		var _ = BeforeEach(func() {
-			config.Set(configDiagnosticMode, true)
-			config.Set(configApidClusterId, testClusterID)
-			_initPlugin(apid.AllServices())
-			apidSnapshotManager = &dummySnapshotManager{}
-			testChangeMan = &dummyChangeManager{
-				pollChangeWithBackoffChan: make(chan bool, 1),
-			}
-			apidChangeManager = testChangeMan
-			apidTokenManager = &dummyTokenManager{}
-			apidInfo = apidInstanceInfo{
-				InstanceID:   testInstanceID,
-				InstanceName: testInstanceName,
-				ClusterID:    testClusterID,
-				LastSnapshot: testLastSnapshot,
-			}
-			updateApidInstanceInfo()
-
-		})
-
-		var _ = AfterEach(func() {
-			config.Set(configDiagnosticMode, false)
-			if wipeDBAferTest {
-				db, err := dataService.DB()
-				Expect(err).NotTo(HaveOccurred())
-				tx, err := db.Begin()
-				_, err = tx.Exec("DELETE FROM APID")
-				Expect(err).NotTo(HaveOccurred())
-				err = tx.Commit()
-				Expect(err).NotTo(HaveOccurred())
-
-			}
-			wipeDBAferTest = true
-			newInstanceID = true
-			isOfflineMode = false
-		})
-
-		It("offline mode should bootstrap from local DB", func(done Done) {
-
-			Expect(apidInfo.LastSnapshot).NotTo(BeEmpty())
-
-			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-
-				if s, ok := event.(*common.Snapshot); ok {
-					// In this test, the changeManager.pollChangeWithBackoff() has not been launched when changeManager closed
-					// This is because the changeManager.pollChangeWithBackoff() in bootstrap() happened after this handler
-					Expect(s.SnapshotInfo).Should(Equal(testLastSnapshot))
-					Expect(s.Tables).To(BeNil())
-					close(done)
-				}
-			})
-			pie := apid.PluginsInitializedEvent{
-				Description: "plugins initialized",
-			}
-			pie.Plugins = append(pie.Plugins, pluginData)
-			postInitPlugins(pie)
-
-		}, 3)
-
-	})
-
-	Context("online mode", func() {
-		var _ = BeforeEach(func() {
-			_initPlugin(apid.AllServices())
-			createManagers()
-		})
-
-		var _ = AfterEach(func() {
-			if wipeDBAferTest {
-				db, err := dataService.DB()
-				Expect(err).NotTo(HaveOccurred())
-				tx, err := db.Begin()
-				_, err = tx.Exec("DELETE FROM APID")
-				Expect(err).NotTo(HaveOccurred())
-				err = tx.Commit()
-				Expect(err).NotTo(HaveOccurred())
-			}
-			wipeDBAferTest = true
-		})
-
-		const expectedDataScopeId1 = "dataScope1"
-		const expectedDataScopeId2 = "dataScope2"
-
-		var initializeContext = func() {
-			testRouter = apid.API().Router()
-			testServer = httptest.NewServer(testRouter)
-
-			// set up mock server
-			mockParms := MockParms{
-				ReliableAPI:  false,
-				ClusterID:    config.GetString(configApidClusterId),
-				TokenKey:     config.GetString(configConsumerKey),
-				TokenSecret:  config.GetString(configConsumerSecret),
-				Scope:        "ert452",
-				Organization: "att",
-				Environment:  "prod",
-			}
-			testMock = Mock(mockParms, testRouter)
-
-			config.Set(configProxyServerBaseURI, testServer.URL)
-			config.Set(configSnapServerBaseURI, testServer.URL)
-			config.Set(configChangeServerBaseURI, testServer.URL)
-		}
-
-		var restoreContext = func() {
-
-			testServer.Close()
-
-			config.Set(configProxyServerBaseURI, dummyConfigValue)
-			config.Set(configSnapServerBaseURI, dummyConfigValue)
-			config.Set(configChangeServerBaseURI, dummyConfigValue)
-
-		}
-
-		It("should succesfully bootstrap from clean slate", func(done Done) {
-			log.Info("Starting sync tests...")
-			var closeDone <-chan bool
-			initializeContext()
-			// do not wipe DB after.  Lets use it
-			wipeDBAferTest = false
-			var lastSnapshot *common.Snapshot
-
-			expectedSnapshotTables := common.ChangeList{
-				Changes: []common.Change{common.Change{Table: "kms_company"},
-					common.Change{Table: "edgex_apid_cluster"},
-					common.Change{Table: "edgex_data_scope"},
-					common.Change{Table: "kms_app_credential"},
-					common.Change{Table: "kms_app_credential_apiproduct_mapper"},
-					common.Change{Table: "kms_developer"},
-					common.Change{Table: "kms_company_developer"},
-					common.Change{Table: "kms_api_product"},
-					common.Change{Table: "kms_app"}},
-			}
-
-			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-				if s, ok := event.(*common.Snapshot); ok {
-
-					Expect(16).To(Equal(len(knownTables)))
-					Expect(changesRequireDDLSync(expectedSnapshotTables)).To(BeFalse())
-
-					lastSnapshot = s
-
-					db, _ := dataService.DBVersion(s.SnapshotInfo)
-					var rowCount int
-					var id string
-
-					err := db.Ping()
-					Expect(err).NotTo(HaveOccurred())
-					numApidClusters, err := db.Query("select distinct count(*) from edgex_apid_cluster;")
-					if err != nil {
-						Fail("Failed to get correct DB")
-					}
-					Expect(true).To(Equal(numApidClusters.Next()))
-					numApidClusters.Scan(&rowCount)
-					Expect(1).To(Equal(rowCount))
-					numApidClusters.Close()
-					apidClusters, err := db.Query("select id from edgex_apid_cluster;")
-					Expect(err).NotTo(HaveOccurred())
-					apidClusters.Next()
-					apidClusters.Scan(&id)
-					Expect(id).To(Equal(expectedClusterId))
-					apidClusters.Close()
-
-					numDataScopes, err := db.Query("select distinct count(*) from edgex_data_scope;")
-					Expect(err).NotTo(HaveOccurred())
-					Expect(true).To(Equal(numDataScopes.Next()))
-					numDataScopes.Scan(&rowCount)
-					Expect(2).To(Equal(rowCount))
-					numDataScopes.Close()
-					dataScopes, err := db.Query("select id from edgex_data_scope;")
-					Expect(err).NotTo(HaveOccurred())
-					dataScopes.Next()
-					dataScopes.Scan(&id)
-					dataScopes.Next()
-
-					if id == expectedDataScopeId1 {
-						dataScopes.Scan(&id)
-						Expect(id).To(Equal(expectedDataScopeId2))
-					} else {
-						dataScopes.Scan(&id)
-						Expect(id).To(Equal(expectedDataScopeId1))
-					}
-					dataScopes.Close()
-
-				} else if cl, ok := event.(*common.ChangeList); ok {
-					closeDone = apidChangeManager.close()
-					// ensure that snapshot switched DB versions
-					Expect(apidInfo.LastSnapshot).To(Equal(lastSnapshot.SnapshotInfo))
-					expectedDB, err := dataService.DBVersion(lastSnapshot.SnapshotInfo)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(getDB() == expectedDB).Should(BeTrue())
-
-					Expect(cl.Changes).To(HaveLen(6))
-
-					var tables []string
-					for _, c := range cl.Changes {
-						tables = append(tables, c.Table)
-						Expect(c.NewRow).ToNot(BeNil())
-
-						var tenantID string
-						c.NewRow.Get("tenant_id", &tenantID)
-						Expect(tenantID).To(Equal("ert452"))
-					}
-
-					Expect(tables).To(ContainElement("kms_app_credential"))
-					Expect(tables).To(ContainElement("kms_app_credential_apiproduct_mapper"))
-					Expect(tables).To(ContainElement("kms_developer"))
-					Expect(tables).To(ContainElement("kms_company_developer"))
-					Expect(tables).To(ContainElement("kms_api_product"))
-					Expect(tables).To(ContainElement("kms_app"))
-
-					go func() {
-						// when close done, all handlers for the first changeList have been executed
-						<-closeDone
-						defer GinkgoRecover()
-						// allow other handler to execute to insert last_sequence
-						var seq string
-						err = getDB().
-							QueryRow("SELECT last_sequence FROM EDGEX_APID_CLUSTER LIMIT 1;").
-							Scan(&seq)
-						Expect(err).NotTo(HaveOccurred())
-						//}
-						Expect(seq).To(Equal(cl.LastSequence))
-
-						restoreContext()
-						close(done)
-					}()
-
-				}
-			})
-			pie := apid.PluginsInitializedEvent{
-				Description: "plugins initialized",
-			}
-			pie.Plugins = append(pie.Plugins, pluginData)
-			postInitPlugins(pie)
-		}, 5)
-
-		It("should bootstrap from local DB if present", func(done Done) {
-
-			var closeDone <-chan bool
-
-			initializeContext()
-			expectedTables := common.ChangeList{
-				Changes: []common.Change{common.Change{Table: "kms_company"},
-					common.Change{Table: "edgex_apid_cluster"},
-					common.Change{Table: "edgex_data_scope"}},
-			}
-			Expect(apidInfo.LastSnapshot).NotTo(BeEmpty())
-
-			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-
-				if s, ok := event.(*common.Snapshot); ok {
-					// In this test, the changeManager.pollChangeWithBackoff() has not been launched when changeManager closed
-					// This is because the changeManager.pollChangeWithBackoff() in bootstrap() happened after this handler
-					closeDone = apidChangeManager.close()
-					go func() {
-						// when close done, all handlers for the first snapshot have been executed
-						<-closeDone
-						//verify that the knownTables array has been properly populated from existing DB
-						Expect(changesRequireDDLSync(expectedTables)).To(BeFalse())
-
-						Expect(s.SnapshotInfo).Should(Equal(apidInfo.LastSnapshot))
-						Expect(s.Tables).To(BeNil())
-
-						restoreContext()
-						close(done)
-					}()
-
-				}
-			})
-			pie := apid.PluginsInitializedEvent{
-				Description: "plugins initialized",
-			}
-			pie.Plugins = append(pie.Plugins, pluginData)
-			postInitPlugins(pie)
-
-		}, 3)
-
-		It("should detect apid_cluster_id change in config yaml", func() {
-			Expect(apidInfo).ToNot(BeNil())
-			Expect(apidInfo.ClusterID).To(Equal("bootstrap"))
-			Expect(apidInfo.InstanceID).ToNot(BeEmpty())
-			previousInstanceId := apidInfo.InstanceID
-
-			config.Set(configApidClusterId, "new value")
-			apidInfo, err := getApidInstanceInfo()
-			Expect(err).NotTo(HaveOccurred())
-			Expect(apidInfo.LastSnapshot).To(BeEmpty())
-			Expect(apidInfo.InstanceID).ToNot(BeEmpty())
-			Expect(apidInfo.InstanceID).ToNot(Equal(previousInstanceId))
-			Expect(apidInfo.ClusterID).To(Equal("new value"))
-		})
-
-		It("should correctly identify non-proper subsets with respect to maps", func() {
-
-			//test b proper subset of a
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
-				[]common.Change{common.Change{Table: "b"}},
-			)).To(BeFalse())
-
-			//test a == b
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
-				[]common.Change{common.Change{Table: "a"}, common.Change{Table: "b"}},
-			)).To(BeFalse())
-
-			//test b superset of a
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
-				[]common.Change{common.Change{Table: "a"}, common.Change{Table: "b"}, common.Change{Table: "c"}},
-			)).To(BeTrue())
-
-			//test b not subset of a
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
-				[]common.Change{common.Change{Table: "c"}},
-			)).To(BeTrue())
-
-			//test a empty
-			Expect(changesHaveNewTables(map[string]bool{},
-				[]common.Change{common.Change{Table: "a"}},
-			)).To(BeTrue())
-
-			//test b empty
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
-				[]common.Change{},
-			)).To(BeFalse())
-
-			//test b nil
-			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true}, nil)).To(BeFalse())
-
-			//test a nil
-			Expect(changesHaveNewTables(nil,
-				[]common.Change{common.Change{Table: "a"}},
-			)).To(BeTrue())
-		}, 3)
-
-		// todo: disabled for now -
-		// there is precondition I haven't been able to track down that breaks this test on occasion
-		XIt("should process a new snapshot when change server requires it", func(done Done) {
-			oldSnap := apidInfo.LastSnapshot
-			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-				defer GinkgoRecover()
-
-				if s, ok := event.(*common.Snapshot); ok {
-					Expect(s.SnapshotInfo).NotTo(Equal(oldSnap))
-					close(done)
-				}
-			})
-			testMock.forceNewSnapshot()
-		})
-
-		It("Verify the Sequence Number Logic works as expected", func() {
-			Expect(getChangeStatus("1.1.1", "1.1.2")).To(Equal(1))
-			Expect(getChangeStatus("1.1.1", "1.2.1")).To(Equal(1))
-			Expect(getChangeStatus("1.2.1", "1.2.1")).To(Equal(0))
-			Expect(getChangeStatus("1.2.1", "1.2.2")).To(Equal(1))
-			Expect(getChangeStatus("2.2.1", "1.2.2")).To(Equal(-1))
-			Expect(getChangeStatus("2.2.1", "2.2.0")).To(Equal(-1))
-		}, 3)
-
-		/*
-		 * XAPID-869, there should not be any panic if received duplicate snapshots during bootstrap
-		 */
-		It("Should be able to handle duplicate snapshot during bootstrap", func() {
-			initializeContext()
-			apidTokenManager = createSimpleTokenManager()
-			apidTokenManager.start()
-			apidSnapshotManager = createSnapShotManager()
-			//events.Listen(ApigeeSyncEventSelector, &handler{})
-
-			scopes := []string{apidInfo.ClusterID}
-			snapshot := &common.Snapshot{}
-			apidSnapshotManager.downloadSnapshot(true, scopes, snapshot)
-			apidSnapshotManager.storeBootSnapshot(snapshot)
-			apidSnapshotManager.storeDataSnapshot(snapshot)
-			restoreContext()
-			<-apidSnapshotManager.close()
-			apidTokenManager.close()
-		}, 3)
-
-		It("Reuse http.Client connection for multiple concurrent requests", func() {
-			var tr *http.Transport
-			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			}))
-			tr = util.Transport(config.GetString(util.ConfigfwdProxyPortURL))
-			tr.MaxIdleConnsPerHost = maxIdleConnsPerHost
-
-			var rspcnt int = 0
-			ch := make(chan *http.Response)
-			client := &http.Client{Transport: tr}
-			for i := 0; i < 2*maxIdleConnsPerHost; i++ {
-				go func(client *http.Client) {
-					req, err := http.NewRequest("GET", server.URL, nil)
-					resp, err := client.Do(req)
-					if err != nil {
-						Fail("Unable to process Client request")
-					}
-					ch <- resp
-					resp.Body.Close()
-
-				}(client)
-			}
-			for {
-				select {
-				case resp := <-ch:
-					Expect(resp.StatusCode).To(Equal(http.StatusOK))
-					if rspcnt >= 2*maxIdleConnsPerHost-1 {
-						return
-					}
-					rspcnt++
-				default:
-				}
-			}
-
-		}, 3)
-
-	})
-})
diff --git a/backoff.go b/backoff.go
deleted file mode 100644
index bad8077..0000000
--- a/backoff.go
+++ /dev/null
@@ -1,98 +0,0 @@
-// Copyright 2017 Google Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package apidApigeeSync
-
-import (
-	"math"
-	"math/rand"
-	"time"
-)
-
-const defaultInitial time.Duration = 200 * time.Millisecond
-const defaultMax time.Duration = 10 * time.Second
-const defaultFactor float64 = 2
-
-type Backoff struct {
-	attempt         int
-	initial, max    time.Duration
-	jitter          bool
-	backoffStrategy func() time.Duration
-}
-
-type ExponentialBackoff struct {
-	Backoff
-	factor float64
-}
-
-func NewExponentialBackoff(initial, max time.Duration, factor float64, jitter bool) *ExponentialBackoff {
-	backoff := &ExponentialBackoff{}
-
-	if initial <= 0 {
-		initial = defaultInitial
-	}
-	if max <= 0 {
-		max = defaultMax
-	}
-
-	if factor <= 0 {
-		factor = defaultFactor
-	}
-
-	backoff.initial = initial
-	backoff.max = max
-	backoff.attempt = 0
-	backoff.factor = factor
-	backoff.jitter = jitter
-	backoff.backoffStrategy = backoff.exponentialBackoffStrategy
-
-	return backoff
-}
-
-func (b *Backoff) Duration() time.Duration {
-	d := b.backoffStrategy()
-	b.attempt++
-	return d
-}
-
-func (b *ExponentialBackoff) exponentialBackoffStrategy() time.Duration {
-
-	initial := float64(b.Backoff.initial)
-	attempt := float64(b.Backoff.attempt)
-	duration := initial * math.Pow(b.factor, attempt)
-
-	if duration > math.MaxInt64 {
-		return b.max
-	}
-	dur := time.Duration(duration)
-
-	if b.jitter {
-		duration = (rand.Float64()*(duration-initial) + initial)
-	}
-
-	if dur > b.max {
-		return b.max
-	}
-
-	log.Debugf("Backing off for %d ms", int64(dur/time.Millisecond))
-	return dur
-}
-
-func (b *Backoff) Reset() {
-	b.attempt = 0
-}
-
-func (b *Backoff) Attempt() int {
-	return b.attempt
-}
diff --git a/change_test.go b/change_test.go
index 511eb82..97d5fb3 100644
--- a/change_test.go
+++ b/change_test.go
@@ -16,144 +16,196 @@
 
 import (
 	"github.com/apid/apid-core"
+	"github.com/apid/apid-core/api"
 	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"net/http"
 	"net/http/httptest"
-	"os"
+	"strconv"
 	"time"
 )
 
+const (
+	expectedInstanceId = "dummy"
+)
+
 var _ = Describe("Change Agent", func() {
 
 	Context("Change Agent Unit Tests", func() {
 
-		var createTestDb = func(sqlfile string, dbId string) common.Snapshot {
-			initDb(sqlfile, "./mockdb_change.sqlite3")
-			file, err := os.Open("./mockdb_change.sqlite3")
-			Expect(err).Should(Succeed())
-			s := common.Snapshot{}
-			err = processSnapshotServerFileResponse(dbId, file, &s)
-			Expect(err).Should(Succeed())
-			return s
-		}
+		Context("utils", func() {
 
-		var initializeContext = func() {
-			testRouter = apid.API().Router()
-			testServer = httptest.NewServer(testRouter)
+			It("should correctly identify non-proper subsets with respect to maps", func() {
+				//test b proper subset of a
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+					[]common.Change{{Table: "b"}},
+				)).To(BeFalse())
 
-			// set up mock server
-			mockParms := MockParms{
-				ReliableAPI:  true,
-				ClusterID:    config.GetString(configApidClusterId),
-				TokenKey:     config.GetString(configConsumerKey),
-				TokenSecret:  config.GetString(configConsumerSecret),
-				Scope:        "ert452",
-				Organization: "att",
-				Environment:  "prod",
-			}
-			testMock = Mock(mockParms, testRouter)
+				//test a == b
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+					[]common.Change{{Table: "a"}, {Table: "b"}},
+				)).To(BeFalse())
 
-			config.Set(configProxyServerBaseURI, testServer.URL)
-			config.Set(configSnapServerBaseURI, testServer.URL)
-			config.Set(configChangeServerBaseURI, testServer.URL)
-			config.Set(configPollInterval, 1*time.Millisecond)
-		}
+				//test b superset of a
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+					[]common.Change{{Table: "a"}, {Table: "b"}, {Table: "c"}},
+				)).To(BeTrue())
 
-		var restoreContext = func() {
+				//test b not subset of a
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+					[]common.Change{{Table: "c"}},
+				)).To(BeTrue())
 
-			testServer.Close()
-			config.Set(configProxyServerBaseURI, dummyConfigValue)
-			config.Set(configSnapServerBaseURI, dummyConfigValue)
-			config.Set(configChangeServerBaseURI, dummyConfigValue)
-			config.Set(configPollInterval, 10*time.Millisecond)
-		}
+				//test a empty
+				Expect(changesHaveNewTables(map[string]bool{},
+					[]common.Change{{Table: "a"}},
+				)).To(BeTrue())
 
-		var _ = BeforeEach(func() {
-			_initPlugin(apid.AllServices())
-			createManagers()
-			event := createTestDb("./sql/init_mock_db.sql", "test_change")
-			processSnapshot(&event)
-			knownTables = extractTablesFromDB(getDB())
-		})
+				//test b empty
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+					[]common.Change{},
+				)).To(BeFalse())
 
-		var _ = AfterEach(func() {
-			restoreContext()
-			if wipeDBAferTest {
-				db, err := dataService.DB()
-				Expect(err).Should(Succeed())
-				tx, err := db.Begin()
-				_, err = tx.Exec("DELETE FROM APID")
-				Expect(err).Should(Succeed())
-				err = tx.Commit()
-				Expect(err).Should(Succeed())
-			}
-			wipeDBAferTest = true
-		})
+				//test b nil
+				Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true}, nil)).To(BeFalse())
 
-		It("test change agent with authorization failure", func() {
-			log.Debug("test change agent with authorization failure")
-			testTokenManager := &dummyTokenManager{make(chan bool)}
-			apidTokenManager = testTokenManager
-			apidTokenManager.start()
-			apidSnapshotManager = &dummySnapshotManager{}
-			initializeContext()
-			testMock.forceAuthFail()
-			wipeDBAferTest = true
-			apidChangeManager.pollChangeWithBackoff()
-			// auth check fails
-			<-testTokenManager.invalidateChan
-			log.Debug("closing")
-			<-apidChangeManager.close()
-		})
-
-		It("test change agent with too old snapshot", func() {
-			log.Debug("test change agent with too old snapshot")
-			testTokenManager := &dummyTokenManager{make(chan bool)}
-			apidTokenManager = testTokenManager
-			apidTokenManager.start()
-			testSnapshotManager := &dummySnapshotManager{make(chan bool)}
-			apidSnapshotManager = testSnapshotManager
-			initializeContext()
-
-			testMock.passAuthCheck()
-			testMock.forceNewSnapshot()
-			wipeDBAferTest = true
-			apidChangeManager.pollChangeWithBackoff()
-			<-testSnapshotManager.downloadCalledChan
-			log.Debug("closing")
-			<-apidChangeManager.close()
-		})
-
-		It("change agent should retry with authorization failure", func(done Done) {
-			log.Debug("change agent should retry with authorization failure")
-			testTokenManager := &dummyTokenManager{make(chan bool)}
-			apidTokenManager = testTokenManager
-			apidTokenManager.start()
-			apidSnapshotManager = &dummySnapshotManager{}
-			initializeContext()
-			testMock.forceAuthFail()
-			testMock.forceNoSnapshot()
-			wipeDBAferTest = true
-
-			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-
-				if _, ok := event.(*common.ChangeList); ok {
-					closeDone := apidChangeManager.close()
-					log.Debug("closing")
-					go func() {
-						// when close done, all handlers for the first snapshot have been executed
-						<-closeDone
-						close(done)
-					}()
-
-				}
+				//test a nil
+				Expect(changesHaveNewTables(nil,
+					[]common.Change{{Table: "a"}},
+				)).To(BeTrue())
 			})
 
-			apidChangeManager.pollChangeWithBackoff()
-			// auth check fails
-			<-testTokenManager.invalidateChan
-		}, 2)
+			It("Compare Sequence Number", func() {
+				Expect(getChangeStatus("1.1.1", "1.1.2")).To(Equal(1))
+				Expect(getChangeStatus("1.1.1", "1.2.1")).To(Equal(1))
+				Expect(getChangeStatus("1.2.1", "1.2.1")).To(Equal(0))
+				Expect(getChangeStatus("1.2.1", "1.2.2")).To(Equal(1))
+				Expect(getChangeStatus("2.2.1", "1.2.2")).To(Equal(-1))
+				Expect(getChangeStatus("2.2.1", "2.2.0")).To(Equal(-1))
+			})
 
+		})
+
+		Context("changeManager", func() {
+			testCount := 0
+			var testChangeMan *pollChangeManager
+			var dummyDbMan *dummyDbManager
+			var dummySnapMan *dummySnapshotManager
+			var dummyTokenMan *dummyTokenManager
+			var testServer *httptest.Server
+			var testRouter apid.Router
+			var testMock *MockServer
+			BeforeEach(func() {
+				testCount++
+				dummyDbMan = &dummyDbManager{
+					knownTables: map[string]bool{
+						"_transicator_metadata":                true,
+						"_transicator_tables":                  true,
+						"attributes":                           true,
+						"edgex_apid_cluster":                   true,
+						"edgex_data_scope":                     true,
+						"kms_api_product":                      true,
+						"kms_app":                              true,
+						"kms_app_credential":                   true,
+						"kms_app_credential_apiproduct_mapper": true,
+						"kms_company":                          true,
+						"kms_company_developer":                true,
+						"kms_deployment":                       true,
+						"kms_developer":                        true,
+						"kms_organization":                     true,
+					},
+					scopes:         []string{"43aef41d"},
+					lastSeqUpdated: make(chan string, 1),
+				}
+				dummySnapMan = &dummySnapshotManager{
+					downloadCalledChan: make(chan bool, 1),
+				}
+				dummyTokenMan = &dummyTokenManager{
+					invalidateChan: make(chan bool, 1),
+				}
+				client := &http.Client{}
+				testChangeMan = createChangeManager(dummyDbMan, dummySnapMan, dummyTokenMan, client)
+				testChangeMan.block = 0
+
+				// create a new API service to have a new router for testing
+				testRouter = api.CreateService().Router()
+				testServer = httptest.NewServer(testRouter)
+				// set up mock server
+				mockParms := MockParms{
+					ReliableAPI:  true,
+					ClusterID:    config.GetString(configApidClusterId),
+					TokenKey:     config.GetString(configConsumerKey),
+					TokenSecret:  config.GetString(configConsumerSecret),
+					Scope:        "",
+					Organization: "att",
+					Environment:  "prod",
+				}
+				apidInfo.ClusterID = expectedClusterId
+				apidInfo.InstanceID = expectedInstanceId
+				testMock = Mock(mockParms, testRouter)
+				config.Set(configProxyServerBaseURI, testServer.URL)
+				config.Set(configSnapServerBaseURI, testServer.URL)
+				config.Set(configChangeServerBaseURI, testServer.URL)
+				config.Set(configPollInterval, 1*time.Millisecond)
+
+				initialBackoffInterval = time.Millisecond
+				testMock.oauthToken = "test_token_" + strconv.Itoa(testCount)
+				dummyTokenMan.token = testMock.oauthToken
+
+			})
+
+			AfterEach(func() {
+				testServer.Close()
+				<-testChangeMan.close()
+				config.Set(configProxyServerBaseURI, dummyConfigValue)
+				config.Set(configSnapServerBaseURI, dummyConfigValue)
+				config.Set(configChangeServerBaseURI, dummyConfigValue)
+				config.Set(configPollInterval, 10*time.Millisecond)
+			})
+
+			It("test change agent with authorization failure", func() {
+				log.Debug("test change agent with authorization failure")
+				testMock.forceAuthFailOnce()
+				testChangeMan.pollChangeWithBackoff()
+				// auth check fails
+				<-dummyTokenMan.invalidateChan
+				log.Debug("closing")
+			})
+
+			It("test change agent with too old snapshot", func() {
+				log.Debug("test change agent with too old snapshot")
+				testMock.passAuthCheck()
+				testMock.forceNewSnapshot()
+				testChangeMan.pollChangeWithBackoff()
+				<-dummySnapMan.downloadCalledChan
+				log.Debug("closing")
+			})
+
+			It("change agent should retry with authorization failure", func() {
+				log.Debug("change agent should retry with authorization failure")
+				testMock.forceAuthFailOnce()
+				testMock.forceNoSnapshot()
+				called := false
+				eventService.ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+					if _, ok := event.(*common.ChangeList); ok {
+						called = true
+					}
+				})
+				testChangeMan.pollChangeWithBackoff()
+				<-dummyTokenMan.invalidateChan
+				Expect(<-dummyDbMan.lastSeqUpdated).Should(Equal(testMock.lastSequenceID()))
+				Expect(called).Should(BeTrue())
+			}, 3)
+
+		})
+
+		Context("offline change manager", func() {
+			It("offline change manager should have no effect", func() {
+				o := &offlineChangeManager{}
+				o.pollChangeWithBackoff()
+				<-o.close()
+			})
+		})
 	})
 })
diff --git a/changes.go b/changes.go
index b78a1d1..6afa827 100644
--- a/changes.go
+++ b/changes.go
@@ -16,34 +16,44 @@
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/apigee-labs/transicator/common"
 	"io/ioutil"
 	"net/http"
 	"net/url"
 	"path"
 	"sort"
+	"strconv"
 	"sync/atomic"
 	"time"
 )
 
-var lastSequence string
-var block string = "45"
-
 type pollChangeManager struct {
 	// 0 for not closed, 1 for closed
 	isClosed *int32
 	// 0 for pollChangeWithBackoff() not launched, 1 for launched
-	isLaunched *int32
-	quitChan   chan bool
+	isLaunched   *int32
+	quitChan     chan bool
+	block        int
+	lastSequence string
+	dbMan        DbManager
+	snapMan      snapshotManager
+	tokenMan     tokenManager
+	client       *http.Client
 }
 
-func createChangeManager() *pollChangeManager {
+func createChangeManager(dbMan DbManager, snapMan snapshotManager, tokenMan tokenManager, client *http.Client) *pollChangeManager {
 	isClosedInt := int32(0)
 	isLaunchedInt := int32(0)
 	return &pollChangeManager{
 		isClosed:   &isClosedInt,
 		quitChan:   make(chan bool),
 		isLaunched: &isLaunchedInt,
+		block:      45,
+		dbMan:      dbMan,
+		snapMan:    snapMan,
+		tokenMan:   tokenMan,
+		client:     client,
 	}
 }
 
@@ -56,7 +66,7 @@
 	finishChan := make(chan bool, 1)
 	//has been closed
 	if atomic.SwapInt32(c.isClosed, 1) == int32(1) {
-		log.Error("pollChangeManager: close() called on a closed pollChangeManager!")
+		log.Warn("pollChangeManager: close() called on a closed pollChangeManager!")
 		go func() {
 			log.Debug("change manager closed")
 			finishChan <- false
@@ -65,11 +75,10 @@
 	}
 	// not launched
 	if atomic.LoadInt32(c.isLaunched) == int32(0) {
-		log.Warn("pollChangeManager: close() called when pollChangeWithBackoff unlaunched! Will wait until pollChangeWithBackoff is launched and then kill it and tokenManager!")
+		log.Warn("pollChangeManager: close() called when pollChangeWithBackoff unlaunched!")
 		go func() {
-			c.quitChan <- true
-			apidTokenManager.close()
-			<-apidSnapshotManager.close()
+			c.tokenMan.close()
+			<-c.snapMan.close()
 			log.Debug("change manager closed")
 			finishChan <- false
 		}()
@@ -79,8 +88,8 @@
 	log.Debug("pollChangeManager: close pollChangeWithBackoff and token manager")
 	go func() {
 		c.quitChan <- true
-		apidTokenManager.close()
-		<-apidSnapshotManager.close()
+		c.tokenMan.close()
+		<-c.snapMan.close()
 		log.Debug("change manager closed")
 		finishChan <- true
 	}()
@@ -120,43 +129,147 @@
 	 * Check to see if we have lastSequence already saved in the DB,
 	 * in which case, it has to be used to prevent re-reading same data
 	 */
-	lastSequence = getLastSequence()
-
+	c.lastSequence = c.dbMan.getLastSequence()
 	for {
 		select {
 		case <-c.quitChan:
 			log.Info("pollChangeAgent; Recevied quit signal to stop polling change server, close token manager")
-			return quitSignalError{}
+			return quitSignalError
 		default:
-			err := c.getChanges(changesUri)
+			scopes, err := c.dbMan.findScopesForId(apidInfo.ClusterID)
 			if err != nil {
-				if _, ok := err.(quitSignalError); ok {
-					log.Debug("pollChangeAgent: consuming the quit signal")
-					<-c.quitChan
-				}
+				return err
+			}
+			r, err := c.getChanges(scopes, changesUri)
+			if err != nil {
+				return err
+			}
+			cl, err := c.parseChangeResp(r)
+			if err != nil {
+				return err
+			}
+			if err = c.emitChangeList(scopes, cl); err != nil {
 				return err
 			}
 		}
 	}
 }
 
-//TODO refactor this method more, split it up
-/* Make a single request to the changeserver to get a changelist */
-func (c *pollChangeManager) getChanges(changesUri *url.URL) error {
-	// if closed
-	if atomic.LoadInt32(c.isClosed) == int32(1) {
-		return quitSignalError{}
+func (c *pollChangeManager) parseChangeResp(r *http.Response) (*common.ChangeList, error) {
+	var err error
+	defer r.Body.Close()
+
+	if r.StatusCode != http.StatusOK {
+		log.Errorf("Get changes request failed with status code: %d", r.StatusCode)
+		switch r.StatusCode {
+		case http.StatusUnauthorized:
+			c.tokenMan.invalidateToken()
+			return nil, authFailError
+
+		case http.StatusNotModified:
+			return nil, nil
+		case http.StatusBadRequest:
+			var apiErr changeServerError
+			var b []byte
+			b, err = ioutil.ReadAll(r.Body)
+			if err != nil {
+				log.Errorf("Unable to read response body: %v", err)
+				return nil, err
+			}
+			err = json.Unmarshal(b, &apiErr)
+			if err != nil {
+				log.Errorf("JSON Response Data not parsable: %s", string(b))
+				return nil, err
+			}
+			if apiErr.Code == "SNAPSHOT_TOO_OLD" {
+				log.Debug("Received SNAPSHOT_TOO_OLD message from change server.")
+				err = apiErr
+			}
+			return nil, err
+		default:
+			log.Errorf("Unknown response code from change server: %v", r.Status)
+			return nil, fmt.Errorf("unknown response code from change server: %v", r.Status)
+		}
 	}
+
+	resp := &common.ChangeList{}
+	err = json.NewDecoder(r.Body).Decode(resp)
+	if err != nil {
+		log.Errorf("JSON Response Data not parsable: %v", err)
+		return nil, err
+	}
+	return resp, nil
+}
+
+func (c *pollChangeManager) emitChangeList(scopes []string, cl *common.ChangeList) error {
+	var err error
+	/*
+	 * If the lastSequence is already newer or the same than what we got via
+	 * cl.LastSequence, Ignore it.
+	 */
+	if c.lastSequence != "" &&
+		getChangeStatus(c.lastSequence, cl.LastSequence) != 1 {
+		return nil
+	}
+
+	if changesRequireDDLSync(c.dbMan.getKnowTables(), cl) {
+		return changeServerError{
+			Code: "DDL changes detected; must get new snapshot",
+		}
+	}
+
+	/* If valid data present, Emit to plugins */
+	if len(cl.Changes) > 0 {
+		if err = c.dbMan.processChangeList(cl); err != nil {
+			log.Errorf("Error in processChangeList: %v", err)
+			return err
+		}
+		/*
+		* Check to see if there was any change in scope. If found, handle it
+		* by getting a new snapshot
+		 */
+		newScopes, err := c.dbMan.findScopesForId(apidInfo.ClusterID)
+		if err != nil {
+			return err
+		}
+		cs := scopeChanged(newScopes, scopes)
+		if cs != nil {
+			return cs
+		}
+		select {
+		case <-time.After(httpTimeout):
+			log.Panic("Timeout. Plugins failed to respond to changes.")
+		case <-eventService.Emit(ApigeeSyncEventSelector, cl):
+		}
+	} else if c.lastSequence == "" { // emit the first changelist anyway
+		select {
+		case <-time.After(httpTimeout):
+			log.Panic("Timeout. Plugins failed to respond to changes.")
+		case <-eventService.Emit(ApigeeSyncEventSelector, cl):
+		}
+	} else {
+		log.Debugf("No Changes detected")
+	}
+
+	err = c.dbMan.updateLastSequence(cl.LastSequence)
+	if err != nil {
+		log.Panicf("Unable to update Sequence in DB. Err {%v}", err)
+	}
+	c.lastSequence = cl.LastSequence
+	return nil
+}
+
+/* Make a single request to the changeserver to get a changelist */
+func (c *pollChangeManager) getChanges(scopes []string, changesUri *url.URL) (*http.Response, error) {
 	log.Debug("polling...")
 
 	/* Find the scopes associated with the config id */
-	scopes := findScopesForId(apidInfo.ClusterID)
 	v := url.Values{}
 
-	blockValue := block
+	blockValue := strconv.Itoa(c.block)
 	/* Sequence added to the query if available */
-	if lastSequence != "" {
-		v.Add("since", lastSequence)
+	if c.lastSequence != "" {
+		v.Add("since", c.lastSequence)
 	} else {
 		blockValue = "0"
 	}
@@ -178,115 +291,16 @@
 
 	/* If error, break the loop, and retry after interval */
 	req, err := http.NewRequest("GET", uri, nil)
-	addHeaders(req)
-	r, err := httpclient.Do(req)
+	addHeaders(req, c.tokenMan.getBearerToken())
+	r, err := c.client.Do(req)
 	if err != nil {
 		log.Errorf("change agent comm error: %s", err)
-		// if closed
-		if atomic.LoadInt32(c.isClosed) == int32(1) {
-			return quitSignalError{}
-		}
-		return err
+		return nil, err
 	}
-	defer r.Body.Close()
-
-	// has been closed
-	if atomic.LoadInt32(c.isClosed) == int32(1) {
-		log.Debugf("getChanges: changeManager has been closed")
-		return quitSignalError{}
-	}
-
-	if r.StatusCode != http.StatusOK {
-		log.Errorf("Get changes request failed with status code: %d", r.StatusCode)
-		switch r.StatusCode {
-		case http.StatusUnauthorized:
-			err = apidTokenManager.invalidateToken()
-			if err != nil {
-				return err
-			}
-			return authFailError{}
-
-		case http.StatusNotModified:
-			return nil
-
-		case http.StatusBadRequest:
-			var apiErr changeServerError
-			var b []byte
-			b, err = ioutil.ReadAll(r.Body)
-			if err != nil {
-				log.Errorf("Unable to read response body: %v", err)
-				return err
-			}
-			err = json.Unmarshal(b, &apiErr)
-			if err != nil {
-				log.Errorf("JSON Response Data not parsable: %s", string(b))
-				return err
-			}
-			if apiErr.Code == "SNAPSHOT_TOO_OLD" {
-				log.Debug("Received SNAPSHOT_TOO_OLD message from change server.")
-				err = apiErr
-			}
-			return err
-		}
-		return nil
-	}
-
-	var resp common.ChangeList
-	err = json.NewDecoder(r.Body).Decode(&resp)
-	if err != nil {
-		log.Errorf("JSON Response Data not parsable: %v", err)
-		return err
-	}
-
-	/*
-	 * If the lastSequence is already newer or the same than what we got via
-	 * resp.LastSequence, Ignore it.
-	 */
-	if lastSequence != "" &&
-		getChangeStatus(lastSequence, resp.LastSequence) != 1 {
-		return nil
-	}
-
-	if changesRequireDDLSync(resp) {
-		return changeServerError{
-			Code: "DDL changes detected; must get new snapshot",
-		}
-	}
-
-	/* If valid data present, Emit to plugins */
-	if len(resp.Changes) > 0 {
-		processChangeList(&resp)
-		select {
-		case <-time.After(httpTimeout):
-			log.Panic("Timeout. Plugins failed to respond to changes.")
-		case <-events.Emit(ApigeeSyncEventSelector, &resp):
-		}
-	} else if lastSequence == "" {
-		select {
-		case <-time.After(httpTimeout):
-			log.Panic("Timeout. Plugins failed to respond to changes.")
-		case <-events.Emit(ApigeeSyncEventSelector, &resp):
-		}
-	} else {
-		log.Debugf("No Changes detected for Scopes: %s", scopes)
-	}
-
-	updateSequence(resp.LastSequence)
-
-	/*
-	 * Check to see if there was any change in scope. If found, handle it
-	 * by getting a new snapshot
-	 */
-	newScopes := findScopesForId(apidInfo.ClusterID)
-	cs := scopeChanged(newScopes, scopes)
-	if cs != nil {
-		return cs
-	}
-
-	return nil
+	return r, nil
 }
 
-func changesRequireDDLSync(changes common.ChangeList) bool {
+func changesRequireDDLSync(knownTables map[string]bool, changes *common.ChangeList) bool {
 	return changesHaveNewTables(knownTables, changes.Changes)
 }
 
@@ -296,10 +310,12 @@
 		log.Debugf("handleChangeServerError: changeManager has been closed")
 		return
 	}
-	if c, ok := err.(changeServerError); ok {
-		log.Debugf("%s. Fetch a new snapshot to sync...", c.Code)
-		apidSnapshotManager.downloadDataSnapshot()
-	} else {
+
+	switch e := err.(type) {
+	case changeServerError:
+		log.Debugf("%s. Fetch a new snapshot to sync...", e.Code)
+		c.snapMan.downloadDataSnapshot()
+	default:
 		log.Debugf("Error connecting to changeserver: %v", err)
 	}
 }
@@ -310,7 +326,7 @@
 func changesHaveNewTables(a map[string]bool, changes []common.Change) bool {
 
 	//nil maps should not be passed in.  Making the distinction between nil map and empty map
-	if a == nil {
+	if len(a) == 0 {
 		log.Warn("Nil map passed to function changesHaveNewTables, may be bug")
 		return true
 	}
@@ -332,24 +348,15 @@
 func getChangeStatus(lastSeq string, currSeq string) int {
 	seqPrev, err := common.ParseSequence(lastSeq)
 	if err != nil {
-		log.Panic("Unable to parse previous sequence string")
+		log.Panicf("Unable to parse previous sequence string: %v", err)
 	}
 	seqCurr, err := common.ParseSequence(currSeq)
 	if err != nil {
-		log.Panic("Unable to parse current sequence string")
+		log.Panicf("Unable to parse current sequence string: %v", err)
 	}
 	return seqCurr.Compare(seqPrev)
 }
 
-func updateSequence(seq string) {
-	lastSequence = seq
-	err := updateLastSequence(seq)
-	if err != nil {
-		log.Panicf("Unable to update Sequence in DB. Err {%v}", err)
-	}
-
-}
-
 /*
  * Returns nil if the two arrays have matching contents
  */
diff --git a/cmd/mockServer/main.go b/cmd/mockServer/main.go
index 486da41..8cfa1ef 100644
--- a/cmd/mockServer/main.go
+++ b/cmd/mockServer/main.go
@@ -19,7 +19,6 @@
 
 	"os"
 
-
 	"github.com/apid/apid-core"
 	"github.com/apid/apid-core/factory"
 	"github.com/apid/apidApigeeSync"
@@ -53,16 +52,16 @@
 	router := apid.API().Router()
 
 	params := apidApigeeSync.MockParms{
-		ReliableAPI:            *reliable,
-		ClusterID:              "cluster",
-		TokenKey:               "key",
-		TokenSecret:            "secret",
-		Scope:                  "scope",
-		Organization:           "org",
-		Environment:            "test",
-		NumDevelopers:          *numDevs,
-		NumDeployments:         *numDeps,
-		BundleURI:              *bundleURI,
+		ReliableAPI:    *reliable,
+		ClusterID:      "cluster",
+		TokenKey:       "key",
+		TokenSecret:    "secret",
+		Scope:          "scope",
+		Organization:   "org",
+		Environment:    "test",
+		NumDevelopers:  *numDevs,
+		NumDeployments: *numDeps,
+		BundleURI:      *bundleURI,
 	}
 
 	log.Printf("Params: %#v\n", params)
diff --git a/data.go b/data.go
index c77fc10..be8dd34 100644
--- a/data.go
+++ b/data.go
@@ -28,27 +28,36 @@
 )
 
 var (
-	unsafeDB apid.DB
-	dbMux    sync.RWMutex
+	dbMux sync.RWMutex
 )
 
-type dataApidCluster struct {
-	ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string
-	Updated, Created                                        string
-}
-
-type dataDataScope struct {
-	ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string
-	Updated, Created                                     string
-}
-
 /*
 This plugin uses 2 databases:
 1. The default DB is used for APID table.
 2. The versioned DB is used for APID_CLUSTER & DATA_SCOPE
 (Currently, the snapshot never changes, but this is future-proof)
 */
-func initDB(db apid.DB) error {
+
+func creatDbManager() *dbManager {
+	return &dbManager{
+		DbMux:       &sync.RWMutex{},
+		knownTables: make(map[string]bool),
+	}
+}
+
+type dbManager struct {
+	Db          apid.DB
+	DbMux       *sync.RWMutex
+	dbVersion   string
+	knownTables map[string]bool
+}
+
+// idempotent call to initialize default DB
+func (dbMan *dbManager) initDB() error {
+	db, err := dataService.DB()
+	if err != nil {
+		return err
+	}
 	tx, err := db.Begin()
 	if err != nil {
 		log.Errorf("initDB(): Unable to get DB tx err: {%v}", err)
@@ -75,24 +84,22 @@
 	return nil
 }
 
-func getDB() apid.DB {
+func (dbMan *dbManager) getDB() apid.DB {
 	dbMux.RLock()
-	db := unsafeDB
-	dbMux.RUnlock()
-	return db
+	defer dbMux.RUnlock()
+	return dbMan.Db
 }
 
-func setDB(db apid.DB) {
+func (dbMan *dbManager) setDB(db apid.DB) {
 	dbMux.Lock()
-	unsafeDB = db
-	dbMux.Unlock()
+	defer dbMux.Unlock()
+	dbMan.Db = db
 }
 
 //TODO if len(rows) > 1000, chunk it up and exec multiple inserts in the txn
-func insert(tableName string, rows []common.Row, txn apid.Tx) bool {
-
+func (dbMan *dbManager) insert(tableName string, rows []common.Row, txn apid.Tx) error {
 	if len(rows) == 0 {
-		return false
+		return fmt.Errorf("no rows")
 	}
 
 	var orderedColumns []string
@@ -101,12 +108,12 @@
 	}
 	sort.Strings(orderedColumns)
 
-	sql := buildInsertSql(tableName, orderedColumns, rows)
+	sql := dbMan.buildInsertSql(tableName, orderedColumns, rows)
 
 	prep, err := txn.Prepare(sql)
 	if err != nil {
-		log.Errorf("INSERT Fail to prepare statement [%s] error=[%v]", sql, err)
-		return false
+		log.Errorf("INSERT Fail to prepare statement %s error=%v", sql, err)
+		return err
 	}
 	defer prep.Close()
 
@@ -123,15 +130,15 @@
 	_, err = prep.Exec(values...)
 
 	if err != nil {
-		log.Errorf("INSERT Fail [%s] values=%v error=[%v]", sql, values, err)
-		return false
+		log.Errorf("INSERT Fail %s values=%v error=%v", sql, values, err)
+		return err
 	}
-	log.Debugf("INSERT Success [%s] values=%v", sql, values)
+	log.Debugf("INSERT Success %s values=%v", sql, values)
 
-	return true
+	return nil
 }
 
-func getValueListFromKeys(row common.Row, pkeys []string) []interface{} {
+func (dbMan *dbManager) getValueListFromKeys(row common.Row, pkeys []string) []interface{} {
 	var values = make([]interface{}, len(pkeys))
 	for i, pkey := range pkeys {
 		if row[pkey] == nil {
@@ -143,52 +150,46 @@
 	return values
 }
 
-func _delete(tableName string, rows []common.Row, txn apid.Tx) bool {
-	pkeys, err := getPkeysForTable(tableName)
+func (dbMan *dbManager) delete(tableName string, rows []common.Row, txn apid.Tx) error {
+	pkeys, err := dbMan.getPkeysForTable(tableName)
 	sort.Strings(pkeys)
 	if len(pkeys) == 0 || err != nil {
-		log.Errorf("DELETE No primary keys found for table. %s", tableName)
-		return false
+		return fmt.Errorf("DELETE No primary keys found for table. %s", tableName)
 	}
 
 	if len(rows) == 0 {
-		log.Errorf("No rows found for table.", tableName)
-		return false
+		return fmt.Errorf("no rows found for table %s", tableName)
 	}
 
-	sql := buildDeleteSql(tableName, rows[0], pkeys)
+	sql := dbMan.buildDeleteSql(tableName, rows[0], pkeys)
 	prep, err := txn.Prepare(sql)
 	if err != nil {
-		log.Errorf("DELETE Fail to prep statement [%s] error=[%v]", sql, err)
-		return false
+		return fmt.Errorf("DELETE Fail to prep statement %s error=%v", sql, err)
 	}
 	defer prep.Close()
 	for _, row := range rows {
-		values := getValueListFromKeys(row, pkeys)
+		values := dbMan.getValueListFromKeys(row, pkeys)
 		// delete prepared statement from existing template statement
 		res, err := txn.Stmt(prep).Exec(values...)
 		if err != nil {
-			log.Errorf("DELETE Fail [%s] values=%v error=[%v]", sql, values, err)
-			return false
+			return fmt.Errorf("DELETE Fail %s values=%v error=%v", sql, values, err)
 		}
 		affected, err := res.RowsAffected()
 		if err == nil && affected != 0 {
-			log.Debugf("DELETE Success [%s] values=%v", sql, values)
+			log.Debugf("DELETE Success %s values=%v", sql, values)
 		} else if err == nil && affected == 0 {
-			log.Errorf("Entry not found [%s] values=%v. Nothing to delete.", sql, values)
-			return false
+			return fmt.Errorf("entry not found %s values=%v, nothing to delete", sql, values)
 		} else {
-			log.Errorf("DELETE Failed [%s] values=%v error=[%v]", sql, values, err)
-			return false
+			return fmt.Errorf("DELETE Failed %s values=%v error=%v", sql, values, err)
 		}
 
 	}
-	return true
+	return nil
 
 }
 
 // Syntax "DELETE FROM Obj WHERE key1=$1 AND key2=$2 ... ;"
-func buildDeleteSql(tableName string, row common.Row, pkeys []string) string {
+func (dbMan *dbManager) buildDeleteSql(tableName string, row common.Row, pkeys []string) string {
 
 	var wherePlaceholders []string
 	i := 1
@@ -210,14 +211,13 @@
 
 }
 
-func update(tableName string, oldRows, newRows []common.Row, txn apid.Tx) bool {
-	pkeys, err := getPkeysForTable(tableName)
+func (dbMan *dbManager) update(tableName string, oldRows, newRows []common.Row, txn apid.Tx) error {
+	pkeys, err := dbMan.getPkeysForTable(tableName)
 	if len(pkeys) == 0 || err != nil {
-		log.Errorf("UPDATE No primary keys found for table.", tableName)
-		return false
+		return fmt.Errorf("UPDATE No primary keys found for table: %v, %v", tableName, err)
 	}
 	if len(oldRows) == 0 || len(newRows) == 0 {
-		return false
+		return fmt.Errorf("UPDATE No old or new rows, table: %v, %v, %v", tableName, oldRows, newRows)
 	}
 
 	var orderedColumns []string
@@ -229,11 +229,10 @@
 	sort.Strings(orderedColumns)
 
 	//build update statement, use arbitrary row as template
-	sql := buildUpdateSql(tableName, orderedColumns, newRows[0], pkeys)
+	sql := dbMan.buildUpdateSql(tableName, orderedColumns, newRows[0], pkeys)
 	prep, err := txn.Prepare(sql)
 	if err != nil {
-		log.Errorf("UPDATE Fail to prep statement [%s] error=[%v]", sql, err)
-		return false
+		return fmt.Errorf("UPDATE Fail to prep statement %s error=%v", sql, err)
 	}
 	defer prep.Close()
 
@@ -265,25 +264,23 @@
 		res, err := txn.Stmt(prep).Exec(values...)
 
 		if err != nil {
-			log.Errorf("UPDATE Fail [%s] values=%v error=[%v]", sql, values, err)
-			return false
+			return fmt.Errorf("UPDATE Fail %s values=%v error=%v", sql, values, err)
 		}
 		numRowsAffected, err := res.RowsAffected()
 		if err != nil {
-			log.Errorf("UPDATE Fail [%s] values=%v error=[%v]", sql, values, err)
-			return false
+			return fmt.Errorf("UPDATE Fail %s values=%v error=%v", sql, values, err)
 		}
 		//delete this once we figure out why tests are failing/not updating
 		log.Debugf("NUM ROWS AFFECTED BY UPDATE: %d", numRowsAffected)
-		log.Debugf("UPDATE Success [%s] values=%v", sql, values)
+		log.Debugf("UPDATE Success %s values=%v", sql, values)
 
 	}
 
-	return true
+	return nil
 
 }
 
-func buildUpdateSql(tableName string, orderedColumns []string, row common.Row, pkeys []string) string {
+func (dbMan *dbManager) buildUpdateSql(tableName string, orderedColumns []string, row common.Row, pkeys []string) string {
 	if row == nil {
 		return ""
 	}
@@ -311,7 +308,7 @@
 }
 
 //precondition: rows.length > 1000, max number of entities for sqlite
-func buildInsertSql(tableName string, orderedColumns []string, rows []common.Row) string {
+func (dbMan *dbManager) buildInsertSql(tableName string, orderedColumns []string, rows []common.Row) string {
 	if len(rows) == 0 {
 		return ""
 	}
@@ -343,13 +340,13 @@
 	return sql
 }
 
-func getPkeysForTable(tableName string) ([]string, error) {
-	db := getDB()
+func (dbMan *dbManager) getPkeysForTable(tableName string) ([]string, error) {
+	db := dbMan.getDB()
 	normalizedTableName := normalizeTableName(tableName)
 	sql := "SELECT columnName FROM _transicator_tables WHERE tableName=$1 AND primaryKey ORDER BY columnName;"
 	rows, err := db.Query(sql, normalizedTableName)
 	if err != nil {
-		log.Errorf("Failed [%s] values=[s%] Error: %v", sql, normalizedTableName, err)
+		log.Errorf("Failed %s values=%s Error: %v", sql, normalizedTableName, err)
 		return nil, err
 	}
 	var columnNames []string
@@ -358,13 +355,15 @@
 		var value string
 		err := rows.Scan(&value)
 		if err != nil {
-			log.Fatal(err)
+			log.Errorf("failed to scan column names: %v", err)
+			return nil, err
 		}
 		columnNames = append(columnNames, value)
 	}
 	err = rows.Err()
 	if err != nil {
-		log.Fatal(err)
+		log.Errorf("failed to scan column names: %v", err)
+		return nil, err
 	}
 	return columnNames, nil
 }
@@ -377,13 +376,11 @@
  * For the given apidConfigId, this function will retrieve all the distinch scopes
  * associated with it. Distinct, because scope is already a collection of the tenants.
  */
-func findScopesForId(configId string) (scopes []string) {
+func (dbMan *dbManager) findScopesForId(configId string) (scopes []string, err error) {
 
 	log.Debugf("findScopesForId: %s", configId)
-
 	var scope sql.NullString
-	db := getDB()
-
+	db := dbMan.getDB()
 	query := `
 		SELECT scope FROM edgex_data_scope WHERE apid_cluster_id = $1
 		UNION
@@ -391,7 +388,6 @@
 		UNION
 		SELECT env_scope FROM edgex_data_scope WHERE apid_cluster_id = $3
 	`
-
 	rows, err := db.Query(query, configId, configId, configId)
 	if err != nil {
 		log.Errorf("Failed to query EDGEX_DATA_SCOPE: %v", err)
@@ -416,9 +412,9 @@
 /*
  * Retrieve SnapshotInfo for the given apidConfigId from apid_config table
  */
-func getLastSequence() (lastSequence string) {
+func (dbMan *dbManager) getLastSequence() (lastSequence string) {
 
-	err := getDB().QueryRow("select last_sequence from EDGEX_APID_CLUSTER LIMIT 1").Scan(&lastSequence)
+	err := dbMan.getDB().QueryRow("select last_sequence from EDGEX_APID_CLUSTER LIMIT 1").Scan(&lastSequence)
 	if err != nil && err != sql.ErrNoRows {
 		log.Panicf("Failed to query EDGEX_APID_CLUSTER: %v", err)
 		return
@@ -432,11 +428,11 @@
  * Persist the last change Id each time a change has been successfully
  * processed by the plugin(s)
  */
-func updateLastSequence(lastSequence string) error {
+func (dbMan *dbManager) updateLastSequence(lastSequence string) error {
 
 	log.Debugf("updateLastSequence: %s", lastSequence)
 
-	tx, err := getDB().Begin()
+	tx, err := dbMan.getDB().Begin()
 	if err != nil {
 		log.Errorf("getApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
 		return err
@@ -454,10 +450,9 @@
 	return err
 }
 
-func getApidInstanceInfo() (info apidInstanceInfo, err error) {
+func (dbMan *dbManager) getApidInstanceInfo() (info apidInstanceInfo, err error) {
 	info.InstanceName = config.GetString(configName)
 	info.ClusterID = config.GetString(configApidClusterId)
-
 	var savedClusterId string
 
 	// always use default database for this
@@ -481,7 +476,7 @@
 		} else {
 			// first start - no row, generate a UUID and store it
 			err = nil
-			newInstanceID = true
+			info.IsNewInstance = true
 			info.InstanceID = util.GenerateUUID()
 
 			log.Debugf("Inserting new apid instance id %s", info.InstanceID)
@@ -489,13 +484,14 @@
 				info.InstanceID, info.ClusterID, "")
 		}
 	} else if savedClusterId != info.ClusterID {
-		log.Debug("Detected apid cluster id change in config.  Apid will start clean")
+		log.Warnf("Detected apid cluster id change in config. %v v.s. %v Apid will start clean.",
+			savedClusterId, info.ClusterID)
 		err = nil
-		newInstanceID = true
+		info.IsNewInstance = true
 		info.InstanceID = util.GenerateUUID()
 
-		_, err = tx.Exec("REPLACE INTO APID (instance_id, apid_cluster_id, last_snapshot_info) VALUES (?,?,?)",
-			info.InstanceID, info.ClusterID, "")
+		_, err = tx.Exec("DELETE FROM APID;")
+
 		info.LastSnapshot = ""
 	}
 	if err = tx.Commit(); err != nil {
@@ -504,8 +500,8 @@
 	return
 }
 
-func updateApidInstanceInfo() error {
-
+func (dbMan *dbManager) updateApidInstanceInfo(instanceId, clusterId, lastSnap string) error {
+	log.Debugf("updateApidInstanceInfo: %v, %v, %v", instanceId, clusterId, lastSnap)
 	// always use default database for this
 	db, err := dataService.DB()
 	if err != nil {
@@ -521,7 +517,7 @@
 		REPLACE
 		INTO APID (instance_id, apid_cluster_id, last_snapshot_info)
 		VALUES (?,?,?)`,
-		apidInfo.InstanceID, apidInfo.ClusterID, apidInfo.LastSnapshot)
+		instanceId, clusterId, lastSnap)
 	if err != nil {
 		log.Errorf("updateApidInstanceInfo: Tx Exec Err: {%v}", err)
 		return err
@@ -536,3 +532,131 @@
 
 	return err
 }
+
+func (dbMan *dbManager) extractTables() (map[string]bool, error) {
+	tables := make(map[string]bool)
+	db := dbMan.getDB()
+	rows, err := db.Query("SELECT DISTINCT tableName FROM _transicator_tables;")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var table sql.NullString
+		if err := rows.Scan(&table); err != nil {
+			return nil, err
+		}
+		log.Debugf("Table %v found in existing db", table)
+		if table.Valid {
+			tables[table.String] = true
+		}
+	}
+	log.Debugf("Extracting table names from existing DB %v", tables)
+	return tables, nil
+}
+
+func (dbMan *dbManager) getKnowTables() map[string]bool {
+	return dbMan.knownTables
+}
+
+func (dbMan *dbManager) processChangeList(changes *common.ChangeList) error {
+
+	tx, err := dbMan.getDB().Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
+
+	for _, change := range changes.Changes {
+		if change.Table == LISTENER_TABLE_APID_CLUSTER {
+			return fmt.Errorf("illegal operation: %s for %s", change.Operation, change.Table)
+		}
+		switch change.Operation {
+		case common.Insert:
+			err = dbMan.insert(change.Table, []common.Row{change.NewRow}, tx)
+		case common.Update:
+			if change.Table == LISTENER_TABLE_DATA_SCOPE {
+				return fmt.Errorf("illegal operation: %s for %s", change.Operation, change.Table)
+			}
+			err = dbMan.update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx)
+		case common.Delete:
+			err = dbMan.delete(change.Table, []common.Row{change.OldRow}, tx)
+		}
+		if err != nil {
+			return err
+		}
+
+	}
+
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("Commit error in processChangeList: %v", err)
+	}
+	return nil
+}
+
+func (dbMan *dbManager) processSnapshot(snapshot *common.Snapshot, isDataSnapshot bool) error {
+
+	var prevDb string
+	if apidInfo.LastSnapshot != "" && apidInfo.LastSnapshot != snapshot.SnapshotInfo {
+		log.Debugf("Release snapshot for {%s}. Switching to version {%s}",
+			apidInfo.LastSnapshot, snapshot.SnapshotInfo)
+		prevDb = apidInfo.LastSnapshot
+	} else {
+		log.Debugf("Process snapshot for version {%s}",
+			snapshot.SnapshotInfo)
+	}
+	db, err := dataService.DBVersion(snapshot.SnapshotInfo)
+	if err != nil {
+		return fmt.Errorf("unable to access database: %v", err)
+	}
+
+	var numApidClusters int
+	tx, err := db.Begin()
+	if err != nil {
+		return fmt.Errorf("unable to open DB txn: {%v}", err.Error())
+	}
+	defer tx.Rollback()
+	err = tx.QueryRow("SELECT COUNT(*) FROM edgex_apid_cluster").Scan(&numApidClusters)
+	if err != nil {
+		return fmt.Errorf("unable to read database: {%s}", err.Error())
+	}
+
+	if numApidClusters != 1 {
+		return fmt.Errorf("illegal state for apid_cluster, must be a single row")
+	}
+
+	_, err = tx.Exec("ALTER TABLE edgex_apid_cluster ADD COLUMN last_sequence text DEFAULT ''")
+	if err != nil && err.Error() != "duplicate column name: last_sequence" {
+		return fmt.Errorf("Unable to create last_sequence column on DB.  Error {%v}", err.Error())
+	}
+
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("error when commit in processSqliteSnapshot: %v", err)
+	}
+
+	//update apid instance info
+	apidInfo.LastSnapshot = snapshot.SnapshotInfo
+	err = dbMan.updateApidInstanceInfo(apidInfo.InstanceID, apidInfo.ClusterID, apidInfo.LastSnapshot)
+	if err != nil {
+		log.Errorf("Unable to update instance info: %v", err)
+		return fmt.Errorf("unable to update instance info: %v", err)
+	}
+
+	dbMan.setDB(db)
+	if isDataSnapshot {
+		dbMan.knownTables, err = dbMan.extractTables()
+		if err != nil {
+			return fmt.Errorf("unable to extract tables: %v", err)
+		}
+	}
+	log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
+
+	// Releases the DB, when the Connection reference count reaches 0.
+	if prevDb != "" {
+		dataService.ReleaseDB(prevDb)
+	}
+	return nil
+}
diff --git a/data_test.go b/data_test.go
index 691e887..abad465 100644
--- a/data_test.go
+++ b/data_test.go
@@ -15,32 +15,54 @@
 package apidApigeeSync
 
 import (
+	"database/sql"
 	"github.com/apid/apid-core"
-	"github.com/apid/apid-core/data"
 	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"io/ioutil"
 	"sort"
 	"strconv"
 )
 
 var _ = Describe("data access tests", func() {
-	testCount := 1
-
-	var _ = BeforeEach(func() {
+	testCount := 0
+	var testDbMan *dbManager
+	var dbVersion string
+	BeforeEach(func() {
+		var testDir string
+		testDir, err := ioutil.TempDir(tmpDir, "data_test")
+		config.Set(configLocalStoragePath, testDir)
+		Expect(err).NotTo(HaveOccurred())
+		testDbMan = creatDbManager()
 		testCount++
-		db, err := dataService.DBVersion("data_test_" + strconv.Itoa(testCount))
+		dbVersion = "data_test_" + strconv.Itoa(testCount)
+		db, err := dataService.DBVersion(dbVersion)
 		Expect(err).Should(Succeed())
-		initDB(db)
-		createBootstrapTables(db)
-		setDB(db)
+		testDbMan.setDB(db)
 	})
 
-	var _ = AfterEach(func() {
-		data.Delete(data.VersionedDBID("common", "data_test_"+strconv.Itoa(testCount)))
+	AfterEach(func() {
+		config.Set(configLocalStoragePath, tmpDir)
 	})
 
-	Context("Update processing", func() {
+	It("check scope changes", func() {
+		newScopes := []string{"foo"}
+		scopes := []string{"bar"}
+		Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
+		newScopes = []string{"foo", "bar"}
+		scopes = []string{"bar"}
+		Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
+		newScopes = []string{"foo"}
+		scopes = []string{"bar", "foo"}
+		Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
+		newScopes = []string{"foo", "bar"}
+		scopes = []string{"bar", "foo"}
+		Expect(scopeChanged(newScopes, scopes)).To(BeNil())
+
+	})
+
+	Context("build Sql", func() {
 		It("unit test buildUpdateSql with single primary key", func() {
 			testRow := common.Row{
 				"id": {
@@ -66,7 +88,7 @@
 			}
 			sort.Strings(orderedColumns)
 
-			result := buildUpdateSql("TEST_TABLE", orderedColumns, testRow, []string{"id"})
+			result := testDbMan.buildUpdateSql("TEST_TABLE", orderedColumns, testRow, []string{"id"})
 			Expect("UPDATE TEST_TABLE SET _change_selector=$1, api_resources=$2, environments=$3, id=$4, tenant_id=$5" +
 				" WHERE id=$6").To(Equal(result))
 		})
@@ -99,403 +121,11 @@
 			}
 			sort.Strings(orderedColumns)
 
-			result := buildUpdateSql("TEST_TABLE", orderedColumns, testRow, []string{"id1", "id2"})
+			result := testDbMan.buildUpdateSql("TEST_TABLE", orderedColumns, testRow, []string{"id1", "id2"})
 			Expect("UPDATE TEST_TABLE SET _change_selector=$1, api_resources=$2, environments=$3, id1=$4, id2=$5, tenant_id=$6" +
 				" WHERE id1=$7 AND id2=$8").To(Equal(result))
 		})
 
-		It("test update with composite primary key", func() {
-			event := &common.ChangeList{}
-
-			//this needs to match what is actually in the DB
-			oldRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "A product for testing Greg",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			newRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "new description",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    oldRow,
-					Operation: 1,
-				},
-			}
-			//insert and assert success
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//create update event
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			//do the update
-			Expect(true).To(Equal(processChangeList(event)))
-			err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-		})
-
-		It("update should succeed if newrow modifies the primary key", func() {
-			event := &common.ChangeList{}
-
-			//this needs to match what is actually in the DB
-			oldRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "A product for testing Greg",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			newRow := common.Row{
-				"id": {
-					Value: "new_id",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "new description",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    oldRow,
-					Operation: 1,
-				},
-			}
-			//insert and assert success
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//create update event
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			//do the update
-			Expect(true).To(Equal(processChangeList(event)))
-			err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='new_id' and description='new description'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-		})
-
-		It("update should succeed if newrow contains fewer fields than oldrow", func() {
-			event := &common.ChangeList{}
-
-			oldRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "A product for testing Greg",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			newRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "new description",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    oldRow,
-					Operation: 1,
-				},
-			}
-			//insert and assert success
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//create update event
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			//do the update
-			Expect(true).To(Equal(processChangeList(event)))
-			err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-		})
-
-		It("update should succeed if oldrow contains fewer fields than newrow", func() {
-			event := &common.ChangeList{}
-
-			oldRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "A product for testing Greg",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			newRow := common.Row{
-				"id": {
-					Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
-				},
-				"api_resources": {
-					Value: "{/**}",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "43aef41d",
-				},
-				"description": {
-					Value: "new description",
-				},
-				"created_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"updated_at": {
-					Value: "2017-03-01 22:50:41.75+00:00",
-				},
-				"_change_selector": {
-					Value: "43aef41d",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    oldRow,
-					Operation: 1,
-				},
-			}
-			//insert and assert success
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//create update event
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    oldRow,
-					NewRow:    newRow,
-					Operation: 2,
-				},
-			}
-
-			//do the update
-			Expect(true).To(Equal(processChangeList(event)))
-			err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-		})
-	})
-
-	Context("Insert processing", func() {
 		It("Properly constructs insert sql for one row", func() {
 			newRow := common.Row{
 				"id": {
@@ -531,7 +161,7 @@
 			sort.Strings(orderedColumns)
 
 			expectedSql := "INSERT INTO api_product(_change_selector,api_resources,created_at,description,environments,id,tenant_id,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)"
-			Expect(expectedSql).To(Equal(buildInsertSql("api_product", orderedColumns, []common.Row{newRow})))
+			Expect(expectedSql).To(Equal(testDbMan.buildInsertSql("api_product", orderedColumns, []common.Row{newRow})))
 		})
 
 		It("Properly constructs insert sql for multiple rows", func() {
@@ -595,275 +225,11 @@
 			sort.Strings(orderedColumns)
 
 			expectedSql := "INSERT INTO api_product(_change_selector,api_resources,created_at,description,environments,id,tenant_id,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8),($9,$10,$11,$12,$13,$14,$15,$16)"
-			Expect(expectedSql).To(Equal(buildInsertSql("api_product", orderedColumns, []common.Row{newRow1, newRow2})))
+			Expect(expectedSql).To(Equal(testDbMan.buildInsertSql("api_product", orderedColumns, []common.Row{newRow1, newRow2})))
 		})
 
-		It("Properly executes insert for a single rows", func() {
-			event := &common.ChangeList{}
-
-			newRow1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow1,
-					Operation: 1,
-				},
-			}
-
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-		})
-
-		It("Properly executed insert for multiple rows", func() {
-			event := &common.ChangeList{}
-
-			newRow1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-			newRow2 := common.Row{
-				"id": {
-					Value: "b",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow1,
-					Operation: 1,
-				},
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow2,
-					Operation: 1,
-				},
-			}
-
-			Expect(true).To(Equal(processChangeList(event)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(2))
-		})
-
-		It("Fails to execute if row does not match existing table schema", func() {
-			event := &common.ChangeList{}
-
-			newRow1 := common.Row{
-				"not_and_id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow1,
-					Operation: 1,
-				},
-			}
-
-			ok := processChangeList(event)
-			Expect(false).To(Equal(ok))
-
-			var nRows int
-			//assert that no extraneous rows were added
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(0))
-		})
-
-		It("Fails to execute at least one row does not match the table schema, even if other rows are valid", func() {
-			event := &common.ChangeList{}
-			newRow1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			newRow2 := common.Row{
-				"not_and_id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			event.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow1,
-					Operation: 1,
-				},
-				{
-					Table:     "kms.api_product",
-					NewRow:    newRow2,
-					Operation: 1,
-				},
-			}
-
-			ok := processChangeList(event)
-			Expect(false).To(Equal(ok))
-		})
-	})
-
-	Context("Delete processing", func() {
 		It("Properly constructs sql prepare for Delete", func() {
+			createBootstrapTables(testDbMan.getDB())
 			row := common.Row{
 				"id": {
 					Value: "new_id",
@@ -891,324 +257,1233 @@
 				},
 			}
 
-			pkeys, err := getPkeysForTable("kms_api_product")
+			pkeys, err := testDbMan.getPkeysForTable("kms_api_product")
 			Expect(err).Should(Succeed())
-			sql := buildDeleteSql("kms_api_product", row, pkeys)
+			sql := testDbMan.buildDeleteSql("kms_api_product", row, pkeys)
 			Expect(sql).To(Equal("DELETE FROM kms_api_product WHERE created_at=$1 AND id=$2 AND tenant_id=$3 AND updated_at=$4"))
 		})
+	})
 
-		It("Verify execute single insert & single delete works", func() {
-			event1 := &common.ChangeList{}
-			event2 := &common.ChangeList{}
-
-			Row1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
-
-			event1.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    Row1,
-					Operation: 1,
-				},
-			}
-			event2.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    Row1,
-					Operation: 3,
-				},
-			}
-
-			Expect(true).To(Equal(processChangeList(event1)))
-			var nRows int
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
-
-			Expect(true).To(Equal(processChangeList(event2)))
-
-			// validate delete
-			err = getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(0))
-
-			// delete again should fail - coz entry will not exist
-			Expect(false).To(Equal(processChangeList(event2)))
+	Context("Process Changelist", func() {
+		BeforeEach(func() {
+			createBootstrapTables(testDbMan.getDB())
 		})
 
-		It("verify multiple insert and single delete works", func() {
-			event1 := &common.ChangeList{}
-			event2 := &common.ChangeList{}
+		Context("Update processing", func() {
+			It("test update with composite primary key", func() {
+				event := &common.ChangeList{}
 
-			Row1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
+				//this needs to match what is actually in the DB
+				oldRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "A product for testing Greg",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-			Row2 := common.Row{
-				"id": {
-					Value: "b",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
+				newRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "new description",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-			event1.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    Row1,
-					Operation: 1,
-				},
-				{
-					Table:     "kms.api_product",
-					NewRow:    Row2,
-					Operation: 1,
-				},
-			}
-			event2.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    Row1,
-					Operation: 3,
-				},
-			}
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    oldRow,
+						Operation: 1,
+					},
+				}
+				//insert and assert success
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
 
-			Expect(true).To(Equal(processChangeList(event1)))
-			var nRows int
-			//verify first row
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+				//create update event
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
 
-			//verify second row
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+				//do the update
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				err = testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
 
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(2))
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
 
-			Expect(true).To(Equal(processChangeList(event2)))
+			})
 
-			//verify second row still exists
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+			It("update should succeed if newrow modifies the primary key", func() {
+				event := &common.ChangeList{}
 
-			// validate delete
-			err = getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+				//this needs to match what is actually in the DB
+				oldRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "A product for testing Greg",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-			// delete again should fail - coz entry will not exist
-			Expect(false).To(Equal(processChangeList(event2)))
-		}, 3)
+				newRow := common.Row{
+					"id": {
+						Value: "new_id",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "new description",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-		It("verify single insert and multiple delete fails", func() {
-			event1 := &common.ChangeList{}
-			event2 := &common.ChangeList{}
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    oldRow,
+						Operation: 1,
+					},
+				}
+				//insert and assert success
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
 
-			Row1 := common.Row{
-				"id": {
-					Value: "a",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
+				//create update event
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
 
-			Row2 := common.Row{
-				"id": {
-					Value: "b",
-				},
-				"api_resources": {
-					Value: "r",
-				},
-				"environments": {
-					Value: "{test}",
-				},
-				"tenant_id": {
-					Value: "t",
-				},
-				"description": {
-					Value: "d",
-				},
-				"created_at": {
-					Value: "c",
-				},
-				"updated_at": {
-					Value: "u",
-				},
-				"_change_selector": {
-					Value: "cs",
-				},
-			}
+				//do the update
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				err = testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='new_id' and description='new description'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
 
-			event1.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					NewRow:    Row1,
-					Operation: 1,
-				},
-			}
-			event2.Changes = []common.Change{
-				{
-					Table:     "kms.api_product",
-					OldRow:    Row1,
-					Operation: 3,
-				},
-				{
-					Table:     "kms.api_product",
-					OldRow:    Row2,
-					Operation: 3,
-				},
-			}
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+			})
 
-			Expect(true).To(Equal(processChangeList(event1)))
-			var nRows int
-			//verify insert
-			err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
-				"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
-				"and _change_selector='cs'").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+			It("update should succeed if newrow contains fewer fields than oldrow", func() {
+				event := &common.ChangeList{}
 
-			//assert that no extraneous rows were added
-			err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(nRows).To(Equal(1))
+				oldRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "A product for testing Greg",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-			Expect(false).To(Equal(processChangeList(event2)))
+				newRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "new description",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
 
-		}, 3)
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    oldRow,
+						Operation: 1,
+					},
+				}
+				//insert and assert success
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//create update event
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
+
+				//do the update
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				err = testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+			})
+
+			It("update should succeed if oldrow contains fewer fields than newrow", func() {
+				event := &common.ChangeList{}
+
+				oldRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "A product for testing Greg",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
+
+				newRow := common.Row{
+					"id": {
+						Value: "87a4bfaa-b3c4-47cd-b6c5-378cdb68610c",
+					},
+					"api_resources": {
+						Value: "{/**}",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "43aef41d",
+					},
+					"description": {
+						Value: "new description",
+					},
+					"created_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"updated_at": {
+						Value: "2017-03-01 22:50:41.75+00:00",
+					},
+					"_change_selector": {
+						Value: "43aef41d",
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    oldRow,
+						Operation: 1,
+					},
+				}
+				//insert and assert success
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//create update event
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    oldRow,
+						NewRow:    newRow,
+						Operation: 2,
+					},
+				}
+
+				//do the update
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				err = testDbMan.getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+			})
+		})
+
+		Context("Insert processing", func() {
+			It("Properly executes insert for a single rows", func() {
+				event := &common.ChangeList{}
+
+				newRow1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow1,
+						Operation: 1,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+			})
+
+			It("Properly executed insert for multiple rows", func() {
+				event := &common.ChangeList{}
+
+				newRow1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+				newRow2 := common.Row{
+					"id": {
+						Value: "b",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow1,
+						Operation: 1,
+					},
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow2,
+						Operation: 1,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(2))
+			})
+
+			It("Fails to execute if row does not match existing table schema", func() {
+				event := &common.ChangeList{}
+
+				newRow1 := common.Row{
+					"not_and_id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow1,
+						Operation: 1,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event)).ShouldNot(Succeed())
+
+				var nRows int
+				//assert that no extraneous rows were added
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(0))
+			})
+
+			It("Fails to execute at least one row does not match the table schema, even if other rows are valid", func() {
+				event := &common.ChangeList{}
+				newRow1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				newRow2 := common.Row{
+					"not_and_id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow1,
+						Operation: 1,
+					},
+					{
+						Table:     "kms.api_product",
+						NewRow:    newRow2,
+						Operation: 1,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event)).ShouldNot(Succeed())
+			})
+		})
+
+		Context("Delete processing", func() {
+
+			It("Verify execute single insert & single delete works", func() {
+				event1 := &common.ChangeList{}
+				event2 := &common.ChangeList{}
+
+				Row1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event1.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    Row1,
+						Operation: 1,
+					},
+				}
+				event2.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    Row1,
+						Operation: 3,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event1)).Should(Succeed())
+				var nRows int
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				Expect(testDbMan.processChangeList(event2)).Should(Succeed())
+
+				// validate delete
+				err = testDbMan.getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(0))
+
+				// delete again should fail - coz entry will not exist
+				Expect(testDbMan.processChangeList(event2)).ShouldNot(Succeed())
+			})
+
+			It("verify multiple insert and single delete works", func() {
+				event1 := &common.ChangeList{}
+				event2 := &common.ChangeList{}
+
+				Row1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				Row2 := common.Row{
+					"id": {
+						Value: "b",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event1.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    Row1,
+						Operation: 1,
+					},
+					{
+						Table:     "kms.api_product",
+						NewRow:    Row2,
+						Operation: 1,
+					},
+				}
+				event2.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    Row1,
+						Operation: 3,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event1)).Should(Succeed())
+				var nRows int
+				//verify first row
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//verify second row
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(2))
+
+				Expect(testDbMan.processChangeList(event2)).Should(Succeed())
+
+				//verify second row still exists
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				// validate delete
+				err = testDbMan.getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				// delete again should fail - coz entry will not exist
+				Expect(testDbMan.processChangeList(event2)).ShouldNot(Succeed())
+			}, 3)
+
+			It("verify single insert and multiple delete fails", func() {
+				event1 := &common.ChangeList{}
+				event2 := &common.ChangeList{}
+
+				Row1 := common.Row{
+					"id": {
+						Value: "a",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				Row2 := common.Row{
+					"id": {
+						Value: "b",
+					},
+					"api_resources": {
+						Value: "r",
+					},
+					"environments": {
+						Value: "{test}",
+					},
+					"tenant_id": {
+						Value: "t",
+					},
+					"description": {
+						Value: "d",
+					},
+					"created_at": {
+						Value: "c",
+					},
+					"updated_at": {
+						Value: "u",
+					},
+					"_change_selector": {
+						Value: "cs",
+					},
+				}
+
+				event1.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						NewRow:    Row1,
+						Operation: 1,
+					},
+				}
+				event2.Changes = []common.Change{
+					{
+						Table:     "kms.api_product",
+						OldRow:    Row1,
+						Operation: 3,
+					},
+					{
+						Table:     "kms.api_product",
+						OldRow:    Row2,
+						Operation: 3,
+					},
+				}
+
+				Expect(testDbMan.processChangeList(event1)).Should(Succeed())
+				var nRows int
+				//verify insert
+				err := testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+					"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
+					"and _change_selector='cs'").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				//assert that no extraneous rows were added
+				err = testDbMan.getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nRows).To(Equal(1))
+
+				Expect(testDbMan.processChangeList(event2)).ShouldNot(Succeed())
+
+			}, 3)
+		})
+
+		Context("ApigeeSync change event", func() {
+
+			Context(LISTENER_TABLE_APID_CLUSTER, func() {
+
+				It("should not change LISTENER_TABLE_APID_CLUSTER", func() {
+					event := common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Insert,
+								Table:     LISTENER_TABLE_APID_CLUSTER,
+							},
+						},
+					}
+					Expect(testDbMan.processChangeList(&event)).NotTo(Succeed())
+
+					event = common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Update,
+								Table:     LISTENER_TABLE_APID_CLUSTER,
+							},
+						},
+					}
+
+					Expect(testDbMan.processChangeList(&event)).NotTo(Succeed())
+				})
+
+			})
+
+			Context("data scopes", func() {
+
+				It("insert event should add", func() {
+					event := common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Insert,
+								Table:     LISTENER_TABLE_DATA_SCOPE,
+								NewRow: common.Row{
+									"id":               &common.ColumnVal{Value: "i"},
+									"apid_cluster_id":  &common.ColumnVal{Value: "a"},
+									"scope":            &common.ColumnVal{Value: "s1"},
+									"org":              &common.ColumnVal{Value: "o"},
+									"env":              &common.ColumnVal{Value: "e"},
+									"created":          &common.ColumnVal{Value: "c"},
+									"created_by":       &common.ColumnVal{Value: "c"},
+									"updated":          &common.ColumnVal{Value: "u"},
+									"updated_by":       &common.ColumnVal{Value: "u"},
+									"_change_selector": &common.ColumnVal{Value: "cs"},
+								},
+							},
+							{
+								Operation: common.Insert,
+								Table:     LISTENER_TABLE_DATA_SCOPE,
+								NewRow: common.Row{
+									"id":               &common.ColumnVal{Value: "j"},
+									"apid_cluster_id":  &common.ColumnVal{Value: "a"},
+									"scope":            &common.ColumnVal{Value: "s2"},
+									"org":              &common.ColumnVal{Value: "o"},
+									"env":              &common.ColumnVal{Value: "e"},
+									"created":          &common.ColumnVal{Value: "c"},
+									"created_by":       &common.ColumnVal{Value: "c"},
+									"updated":          &common.ColumnVal{Value: "u"},
+									"updated_by":       &common.ColumnVal{Value: "u"},
+									"_change_selector": &common.ColumnVal{Value: "cs"},
+								},
+							},
+						},
+					}
+
+					testDbMan.processChangeList(&event)
+
+					count := 0
+					id := sql.NullString{}
+					rows, err := testDbMan.getDB().Query(`
+				SELECT scope FROM EDGEX_DATA_SCOPE`)
+					Expect(err).NotTo(HaveOccurred())
+					defer rows.Close()
+					for rows.Next() {
+						count++
+						Expect(rows.Scan(&id)).Should(Succeed())
+						Expect(id.String).Should(Equal("s" + strconv.Itoa(count)))
+					}
+
+					Expect(count).To(Equal(2))
+				})
+
+				It("delete event should delete", func() {
+					event := common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Insert,
+								Table:     LISTENER_TABLE_DATA_SCOPE,
+								NewRow: common.Row{
+									"id":               &common.ColumnVal{Value: "i"},
+									"apid_cluster_id":  &common.ColumnVal{Value: "a"},
+									"scope":            &common.ColumnVal{Value: "s"},
+									"org":              &common.ColumnVal{Value: "o"},
+									"env":              &common.ColumnVal{Value: "e"},
+									"created":          &common.ColumnVal{Value: "c"},
+									"created_by":       &common.ColumnVal{Value: "c"},
+									"updated":          &common.ColumnVal{Value: "u"},
+									"updated_by":       &common.ColumnVal{Value: "u"},
+									"_change_selector": &common.ColumnVal{Value: "cs"},
+								},
+							},
+						},
+					}
+
+					testDbMan.processChangeList(&event)
+
+					event = common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Delete,
+								Table:     LISTENER_TABLE_DATA_SCOPE,
+								OldRow:    event.Changes[0].NewRow,
+							},
+						},
+					}
+
+					testDbMan.processChangeList(&event)
+
+					var nRows int
+					err := testDbMan.getDB().QueryRow("SELECT count(id) FROM EDGEX_DATA_SCOPE").Scan(&nRows)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(nRows).To(BeZero())
+				})
+
+				It("update event should panic for data scopes table", func() {
+					event := common.ChangeList{
+						LastSequence: "test",
+						Changes: []common.Change{
+							{
+								Operation: common.Update,
+								Table:     LISTENER_TABLE_DATA_SCOPE,
+							},
+						},
+					}
+
+					Expect(testDbMan.processChangeList(&event)).ToNot(Succeed())
+				})
+
+			})
+		})
+
 	})
+
+	Context("Process Snapshot", func() {
+		initTestDb := func(sqlFile string, dbMan *dbManager) common.Snapshot {
+			stmts, err := ioutil.ReadFile(sqlFile)
+			Expect(err).Should(Succeed())
+			Expect(testDbMan.getDB().Exec(string(stmts))).ShouldNot(BeNil())
+			Expect(testDbMan.initDB()).Should(Succeed())
+			return common.Snapshot{
+				SnapshotInfo: dbVersion,
+			}
+		}
+
+		AfterEach(func() {
+
+		})
+
+		It("should fail if more than one apid_cluster rows", func() {
+			event := initTestDb("./sql/init_listener_test_duplicate_apids.sql", testDbMan)
+			Expect(testDbMan.processSnapshot(&event, true)).ToNot(Succeed())
+		})
+
+		It("should process a valid Snapshot", func() {
+			config.Set(configApidClusterId, "a")
+			apidInfo.ClusterID = "a"
+			event := initTestDb("./sql/init_listener_test_valid_snapshot.sql", testDbMan)
+			Expect(testDbMan.processSnapshot(&event, true)).Should(Succeed())
+
+			info, err := testDbMan.getApidInstanceInfo()
+			Expect(err).Should(Succeed())
+			Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo))
+			Expect(info.IsNewInstance).To(BeFalse())
+			Expect(dataService.DBVersion(event.SnapshotInfo)).Should(Equal(testDbMan.getDB()))
+
+			// apid Cluster
+			id := &sql.NullString{}
+			Expect(testDbMan.getDB().QueryRow(`SELECT id FROM EDGEX_APID_CLUSTER`).
+				Scan(id)).Should(Succeed())
+			Expect(id.Valid).Should(BeTrue())
+			Expect(id.String).To(Equal("i"))
+
+			// Data Scope
+			env := &sql.NullString{}
+			count := 0
+			rows, err := testDbMan.getDB().Query(`SELECT env FROM EDGEX_DATA_SCOPE`)
+			Expect(err).Should(Succeed())
+			defer rows.Close()
+			for rows.Next() {
+				count++
+				rows.Scan(&env)
+				Expect(env.Valid).Should(BeTrue())
+				Expect(env.String).To(Equal("e" + strconv.Itoa(count)))
+			}
+			Expect(count).To(Equal(3))
+
+			//find scopes for Id
+			scopes, err := testDbMan.findScopesForId("a")
+			Expect(err).Should(Succeed())
+			Expect(len(scopes)).To(Equal(6))
+			expectedScopes := []string{"s1", "s2", "org_scope_1", "env_scope_1", "env_scope_2", "env_scope_3"}
+			sort.Strings(scopes)
+			sort.Strings(expectedScopes)
+			Expect(scopes).Should(Equal(expectedScopes))
+		})
+
+		It("should detect clusterid change", func() {
+			Expect(testDbMan.initDB()).Should(Succeed())
+			testDbMan.updateApidInstanceInfo("a", "b", "c")
+			config.Set(configApidClusterId, "d")
+
+			info, err := testDbMan.getApidInstanceInfo()
+			Expect(err).Should(Succeed())
+			Expect(info.LastSnapshot).To(BeZero())
+			Expect(info.IsNewInstance).To(BeTrue())
+		})
+	})
+
 })
 
 func createBootstrapTables(db apid.DB) {
 	tx, err := db.Begin()
 	Expect(err).To(Succeed())
 	//all tests in this file operate on the api_product table.  Create the necessary tables for this here
-	tx.Exec("CREATE TABLE _transicator_tables " +
-		"(tableName varchar not null, columnName varchar not null, " +
-		"typid integer, primaryKey bool);")
-	tx.Exec("DELETE from _transicator_tables")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','id',2950,1)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','tenant_id',1043,1)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','description',1043,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','api_resources',1015,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','approval_type',1043,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','scopes',1015,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','proxies',1015,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','environments',1015,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_at',1114,1)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_by',1043,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_at',1114,1)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_by',1043,0)")
-	tx.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','_change_selector',1043,0)")
-
-	tx.Exec("CREATE TABLE kms_api_product (id text,tenant_id text,name text, description text, " +
-		"api_resources text,approval_type text,scopes text,proxies text, environments text," +
-		"created_at blob, created_by text,updated_at blob,updated_by text,_change_selector text, " +
-		"primary key (id,tenant_id,created_at,updated_at));")
-	tx.Exec("DELETE from kms_api_product")
-	err = tx.Commit()
+	_, err = tx.Exec(`CREATE TABLE _transicator_tables
+	(tableName varchar not null, columnName varchar not null, typid integer, primaryKey bool);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','id',2950,1);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','tenant_id',1043,1);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','name',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','display_name',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','description',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','api_resources',1015,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','approval_type',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','scopes',1015,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','proxies',1015,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','environments',1015,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','quota',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','quota_time_unit',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','quota_interval',23,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','created_at',1114,1);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','created_by',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','updated_at',1114,1);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','updated_by',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('kms_api_product','_change_selector',1043,0);
+	CREATE TABLE "kms_api_product" (id text,tenant_id text,name text,display_name text,description text,api_resources text,approval_type text,scopes text,proxies text,environments text,quota text,quota_time_unit text,quota_interval integer,created_at blob,created_by text,updated_at blob,updated_by text,_change_selector text, primary key (id,tenant_id,created_at,updated_at));
+	`)
 	Expect(err).To(Succeed())
+	_, err = tx.Exec(`
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','id',1043,1);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','apid_cluster_id',1043,1);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','scope',25,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','org',25,1);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','env',25,1);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','org_scope',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','env_scope',1043,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','created',1114,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','created_by',25,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','updated',1114,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','updated_by',25,0);
+	INSERT INTO "_transicator_tables" VALUES('edgex_data_scope','_change_selector',25,1);
+	CREATE TABLE "edgex_data_scope" (id text,apid_cluster_id text,scope text,org text,env text,org_scope text,
+	env_scope text,created blob,created_by text,updated blob,updated_by text,_change_selector text,
+	primary key (id,apid_cluster_id,apid_cluster_id,org,env,_change_selector));
+	`)
+	Expect(err).To(Succeed())
+
+	Expect(tx.Commit()).To(Succeed())
 }
diff --git a/dockertests/apid_config.yaml b/dockertests/apid_config.yaml
new file mode 100644
index 0000000..e4cd751
--- /dev/null
+++ b/dockertests/apid_config.yaml
@@ -0,0 +1,9 @@
+apigeesync_instance_name: SQLLITAPID
+apigeesync_snapshot_server_base: http://localhost:9001/
+apigeesync_change_server_base: http://localhost:9000/
+apigeesync_snapshot_proto: sqlite
+log_level: Debug
+apigeesync_consumer_key: 33f39JNLosF1mDOXJoCfbauchVzPrGrl
+apigeesync_consumer_secret: LAolGShAx6H3vfNF
+apigeesync_cluster_id: 4c6bb536-0d64-43ca-abae-17c08f1a7e58
+local_storage_path: /Users/haoming/go/src/github.com/apid/apidApigeeSync/tmp/sqlite
diff --git a/dockertests/dockerSetup.sh b/dockertests/dockerSetup.sh
index 8df2b5d..6ded35d 100755
--- a/dockertests/dockerSetup.sh
+++ b/dockertests/dockerSetup.sh
@@ -25,7 +25,6 @@
 TEST_PG_BASE=postgres://postgres:changeme@$DOCKER_IP:5432
 TEST_PG_URL=postgres://postgres:changeme@$DOCKER_IP:5432/edgex
 echo ${TEST_PG_URL}
-
 export APIGEE_SYNC_DOCKER_PG_URL=${TEST_PG_URL}
 export APIGEE_SYNC_DOCKER_IP=${DOCKER_IP}
 
@@ -47,8 +46,13 @@
 ssname=apidSync_test_ss
 csname=apidSync_test_cs
 
+# setup docker network
+docker network rm apidApigeeSync-docker-test || true
+docker network create apidApigeeSync-docker-test
+DOCKER_PG_URL=postgres://postgres:changeme@$pgname:5432/edgex
+echo $DOCKER_PG_URL
 # run PG
-docker run --name ${pgname} -p 5432:5432 -d -e POSTGRES_PASSWORD=changeme apigeelabs/transicator-postgres
+docker run --name ${pgname} -p 5432:5432 --network=apidApigeeSync-docker-test -d -e POSTGRES_PASSWORD=changeme apigeelabs/transicator-postgres
 
 # Wait for PG to be up -- it takes a few seconds
 while `true`
@@ -67,8 +71,8 @@
 psql -f ${WORK_DIR}/dockertests/user-setup.sql ${TEST_PG_URL}
 
 # run SS and CS
-docker run --name ${ssname} -d -p 9001:9001 apigeelabs/transicator-snapshot -p 9001 -u ${TEST_PG_URL}
-docker run --name ${csname} -d -p 9000:9000 apigeelabs/transicator-changeserver -p 9000 -u ${TEST_PG_URL} -s testslot
+docker run --name ${ssname} -d -p 9001:9001 --network=apidApigeeSync-docker-test apigeelabs/transicator-snapshot -p 9001 -u ${DOCKER_PG_URL}
+docker run --name ${csname} -d -p 9000:9000 --network=apidApigeeSync-docker-test apigeelabs/transicator-changeserver -p 9000 -u ${DOCKER_PG_URL} -s testslot
 
 # Wait for SS to be up
 while `true`
diff --git a/dockertests/docker_test.go b/dockertests/docker_test.go
index e8fc25d..631e17a 100644
--- a/dockertests/docker_test.go
+++ b/dockertests/docker_test.go
@@ -18,6 +18,7 @@
 	"encoding/json"
 	"github.com/apid/apid-core"
 	"github.com/apid/apid-core/factory"
+	"github.com/apid/apidApigeeSync"
 	_ "github.com/apid/apidApigeeSync"
 	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
@@ -77,7 +78,7 @@
 	config.Set(configProxyServerBaseURI, testServer.URL)
 
 	// init plugin
-	apid.RegisterPlugin(initPlugin)
+	apid.RegisterPlugin(initPlugin, apidApigeeSync.PluginData)
 	apid.InitializePlugins("dockerTest")
 
 	<-initDone
@@ -263,6 +264,8 @@
 		updated:        t,
 		updatedBy:      testInitUser,
 		changeSelector: clusterId,
+		orgScope:       "org1",
+		envScope:       "env1",
 	}
 
 	bf := bundleConfigData{
@@ -335,7 +338,7 @@
 type newTableHandler struct {
 	targetTablename string
 	done            Done
-	verifyFunc      func (string, apid.DB)
+	verifyFunc      func(string, apid.DB)
 }
 
 func (n *newTableHandler) Handle(event apid.Event) {
diff --git a/dockertests/management_pg.go b/dockertests/management_pg.go
index bbe28e9..77bdf67 100644
--- a/dockertests/management_pg.go
+++ b/dockertests/management_pg.go
@@ -90,9 +90,11 @@
 			created_by,
 			updated,
 			updated_by,
-			_change_selector
+			_change_selector,
+			org_scope,
+			env_scope
 			)
-			VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`)
+			VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)`)
 	if err != nil {
 		return err
 	}
@@ -108,6 +110,8 @@
 		ds.updated,
 		ds.updatedBy,
 		ds.changeSelector,
+		ds.orgScope,
+		ds.envScope,
 	)
 
 	return err
diff --git a/dockertests/master-schema.sql b/dockertests/master-schema.sql
index 756ff62..00f60d0 100644
--- a/dockertests/master-schema.sql
+++ b/dockertests/master-schema.sql
@@ -45,6 +45,8 @@
     updated timestamp without time zone,
     updated_by text,
     _change_selector text,
+    org_scope character varying(36) NOT NULL,
+    env_scope character varying(36) NOT NULL,
     CONSTRAINT data_scope_pkey PRIMARY KEY (id),
     CONSTRAINT data_scope_apid_cluster_id_fk FOREIGN KEY (apid_cluster_id)
           REFERENCES apid_cluster (id)
diff --git a/dockertests/pg_table_data.go b/dockertests/pg_table_data.go
index 0a1aecd..854e587 100644
--- a/dockertests/pg_table_data.go
+++ b/dockertests/pg_table_data.go
@@ -42,6 +42,8 @@
 	updated        time.Time
 	updatedBy      string
 	changeSelector string
+	orgScope       string
+	envScope       string
 }
 
 /* FOREIGN KEY (data_scope_id)
diff --git a/init.go b/init.go
index b7f470e..f381ca4 100644
--- a/init.go
+++ b/init.go
@@ -15,7 +15,6 @@
 package apidApigeeSync
 
 import (
-	"encoding/json"
 	"fmt"
 	"net/http"
 	"os"
@@ -39,7 +38,8 @@
 	// special value - set by ApigeeSync, not taken from configuration
 	configApidInstanceID = "apigeesync_apid_instance_id"
 	// This will not be needed once we have plugin handling tokens.
-	configBearerToken = "apigeesync_bearer_token"
+	configBearerToken      = "apigeesync_bearer_token"
+	configLocalStoragePath = "local_storage_path"
 )
 
 const (
@@ -48,17 +48,12 @@
 
 var (
 	/* All set during plugin initialization */
-	log                 apid.LogService
-	config              apid.ConfigService
-	dataService         apid.DataService
-	events              apid.EventsService
-	apidInfo            apidInstanceInfo
-	newInstanceID       bool
-	apidTokenManager    tokenManager
-	apidChangeManager   changeManager
-	apidSnapshotManager snapShotManager
-	httpclient          *http.Client
-	isOfflineMode       bool
+	log          apid.LogService
+	config       apid.ConfigService
+	dataService  apid.DataService
+	eventService apid.EventsService
+	apiService   apid.APIService
+	apidInfo     apidInstanceInfo
 
 	/* Set during post plugin initialization
 	 * set this as a default, so that it's guaranteed to be valid even if postInitPlugins isn't called
@@ -68,6 +63,7 @@
 
 type apidInstanceInfo struct {
 	InstanceID, InstanceName, ClusterID, LastSnapshot string
+	IsNewInstance                                     bool
 }
 
 type pluginDetail struct {
@@ -76,7 +72,7 @@
 }
 
 func init() {
-	apid.RegisterPlugin(initPlugin, pluginData)
+	apid.RegisterPlugin(initPlugin, PluginData)
 }
 
 func initConfigDefaults() {
@@ -93,59 +89,7 @@
 	log.Debugf("Using %s as display name", config.GetString(configName))
 }
 
-func initVariables() error {
-
-	var tr *http.Transport
-
-	tr = util.Transport(config.GetString(util.ConfigfwdProxyPortURL))
-	tr.MaxIdleConnsPerHost = maxIdleConnsPerHost
-
-	httpclient = &http.Client{
-		Transport: tr,
-		Timeout:   httpTimeout,
-		CheckRedirect: func(req *http.Request, _ []*http.Request) error {
-			req.Header.Set("Authorization", "Bearer "+apidTokenManager.getBearerToken())
-			return nil
-		},
-	}
-
-	// set up default database
-	db, err := dataService.DB()
-	if err != nil {
-		return fmt.Errorf("Unable to access DB: %v", err)
-	}
-	err = initDB(db)
-	if err != nil {
-		return fmt.Errorf("Unable to access DB: %v", err)
-	}
-	setDB(db)
-
-	apidInfo, err = getApidInstanceInfo()
-	if err != nil {
-		return fmt.Errorf("Unable to get apid instance info: %v", err)
-	}
-
-	if config.IsSet(configApidInstanceID) {
-		log.Warnf("ApigeeSync plugin overriding %s.", configApidInstanceID)
-	}
-	config.Set(configApidInstanceID, apidInfo.InstanceID)
-
-	return nil
-}
-
-func createManagers() {
-	if isOfflineMode {
-		apidSnapshotManager = &offlineSnapshotManager{}
-		apidChangeManager = &offlineChangeManager{}
-	} else {
-		apidSnapshotManager = createSnapShotManager()
-		apidChangeManager = createChangeManager()
-	}
-
-	apidTokenManager = createSimpleTokenManager()
-}
-
-func checkForRequiredValues() error {
+func checkForRequiredValues(isOfflineMode bool) error {
 	required := []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret}
 	if !isOfflineMode {
 		required = append(required, configSnapServerBaseURI, configChangeServerBaseURI)
@@ -153,12 +97,12 @@
 	// check for required values
 	for _, key := range required {
 		if !config.IsSet(key) {
-			return fmt.Errorf("Missing required config value: %s", key)
+			return fmt.Errorf("missing required config value: %s", key)
 		}
 	}
 	proto := config.GetString(configSnapshotProtocol)
 	if proto != "sqlite" {
-		return fmt.Errorf("Illegal value for %s. Only currently supported snashot protocol is sqlite", configSnapshotProtocol)
+		return fmt.Errorf("illegal value for %s. Only currently supported snashot protocol is sqlite", configSnapshotProtocol)
 	}
 
 	return nil
@@ -168,90 +112,98 @@
 	log = logger
 }
 
-/* initialization */
-func _initPlugin(services apid.Services) error {
-	log.Debug("start init")
+func initManagers(isOfflineMode bool) (*listenerManager, *ApiManager, error) {
+	// check for forward proxy
+	var tr *http.Transport
+	tr = util.Transport(config.GetString(util.ConfigfwdProxyPortURL))
+	tr.MaxIdleConnsPerHost = maxIdleConnsPerHost
 
-	config = services.Config()
-	initConfigDefaults()
-
-	if config.GetBool(configDiagnosticMode) {
-		log.Warn("Diagnostic mode: will not download changelist and snapshots!")
-		isOfflineMode = true
-	}
-
-	err := checkForRequiredValues()
+	apidDbManager := creatDbManager()
+	db, err := dataService.DB()
 	if err != nil {
-		return err
+		return nil, nil, fmt.Errorf("unable to access DB: %v", err)
 	}
-
-	err = initVariables()
+	apidDbManager.setDB(db)
+	err = apidDbManager.initDB()
 	if err != nil {
-		return err
+		return nil, nil, fmt.Errorf("unable to access DB: %v", err)
 	}
 
-	return nil
+	apidInfo, err = apidDbManager.getApidInstanceInfo()
+	if err != nil {
+		return nil, nil, fmt.Errorf("unable to get apid instance info: %v", err)
+	}
+
+	if config.IsSet(configApidInstanceID) {
+		log.Warnf("ApigeeSync plugin overriding %s.", configApidInstanceID)
+	}
+	config.Set(configApidInstanceID, apidInfo.InstanceID)
+
+	apidTokenManager := createApidTokenManager(apidInfo.IsNewInstance)
+	var snapMan snapshotManager
+	var apidChangeManager changeManager
+
+	if isOfflineMode {
+		snapMan = &offlineSnapshotManager{
+			dbMan: apidDbManager,
+		}
+		apidChangeManager = &offlineChangeManager{}
+	} else {
+		httpClient := &http.Client{
+			Transport: tr,
+			Timeout:   httpTimeout,
+			CheckRedirect: func(req *http.Request, _ []*http.Request) error {
+				req.Header.Set("Authorization", "Bearer "+apidTokenManager.getBearerToken())
+				return nil
+			},
+		}
+		snapMan = createSnapShotManager(apidDbManager, apidTokenManager, httpClient)
+		apidChangeManager = createChangeManager(apidDbManager, snapMan, apidTokenManager, httpClient)
+	}
+
+	listenerMan := &listenerManager{
+		changeMan:     apidChangeManager,
+		snapMan:       snapMan,
+		tokenMan:      apidTokenManager,
+		isOfflineMode: isOfflineMode,
+	}
+
+	apiMan := &ApiManager{
+		endpoint: tokenEndpoint,
+		tokenMan: apidTokenManager,
+	}
+	return listenerMan, apiMan, nil
 }
 
 func initPlugin(services apid.Services) (apid.PluginData, error) {
 	SetLogger(services.Log().ForModule("apigeeSync"))
 	dataService = services.Data()
-	events = services.Events()
+	eventService = services.Events()
+	apiService = services.API()
+	log.Debug("start init")
+	config = services.Config()
+	initConfigDefaults()
 
-	err := _initPlugin(services)
+	isOfflineMode := false
+	if config.GetBool(configDiagnosticMode) {
+		log.Warn("Diagnostic mode: will not download changelist and snapshots!")
+		isOfflineMode = true
+	}
+
+	err := checkForRequiredValues(isOfflineMode)
 	if err != nil {
-		return pluginData, err
+		return PluginData, err
 	}
+	if err != nil {
+		return PluginData, err
+	}
+	listenerMan, apiMan, err := initManagers(isOfflineMode)
+	if err != nil {
+		return PluginData, err
+	}
+	listenerMan.init()
+	apiMan.InitAPI(apiService)
 
-	createManagers()
-
-	/* This callback function will get called once all the plugins are
-	 * initialized (not just this plugin). This is needed because,
-	 * downloadSnapshots/changes etc have to begin to be processed only
-	 * after all the plugins are initialized
-	 */
-	events.ListenOnceFunc(apid.SystemEventsSelector, postInitPlugins)
-
-	InitAPI(services)
 	log.Debug("end init")
-
-	return pluginData, nil
-}
-
-// Plugins have all initialized, gather their info and start the ApigeeSync downloads
-func postInitPlugins(event apid.Event) {
-	var plinfoDetails []pluginDetail
-	if pie, ok := event.(apid.PluginsInitializedEvent); ok {
-		/*
-		 * Store the plugin details in the heap. Needed during
-		 * Bearer token generation request.
-		 */
-		for _, plugin := range pie.Plugins {
-			name := plugin.Name
-			version := plugin.Version
-			if schemaVersion, ok := plugin.ExtraData["schemaVersion"].(string); ok {
-				inf := pluginDetail{
-					Name:          name,
-					SchemaVersion: schemaVersion}
-				plinfoDetails = append(plinfoDetails, inf)
-				log.Debugf("plugin %s is version %s, schemaVersion: %s", name, version, schemaVersion)
-			}
-		}
-		if plinfoDetails == nil {
-			log.Panicf("No Plugins registered!")
-		}
-
-		pgInfo, err := json.Marshal(plinfoDetails)
-		if err != nil {
-			log.Panicf("Unable to marshal plugin data: %v", err)
-		}
-		apidPluginDetails = string(pgInfo[:])
-
-		log.Debug("start post plugin init")
-
-		apidTokenManager.start()
-		go bootstrap()
-
-		log.Debug("Done post plugin init")
-	}
+	return PluginData, nil
 }
diff --git a/init_test.go b/init_test.go
index 03e5450..bdd9fc7 100644
--- a/init_test.go
+++ b/init_test.go
@@ -18,36 +18,78 @@
 	"github.com/apid/apid-core"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"net/http"
+	"strconv"
 )
 
 var _ = Describe("init", func() {
-	var _ = BeforeEach(func() {
-		_initPlugin(apid.AllServices())
+	testCount := 0
+	BeforeEach(func() {
+		testCount++
 	})
 
 	Context("Apid Instance display name", func() {
+		AfterEach(func() {
+			apiService = apid.API()
+		})
 
-		It("should be hostname by default", func() {
-			log.Info("Starting init tests...")
+		It("init should register listener", func() {
+			me := &mockEvent{
+				listenerMap: make(map[apid.EventSelector]apid.EventHandlerFunc),
+			}
+			ma := &mockApi{
+				handleMap: make(map[string]http.HandlerFunc),
+			}
+			ms := &mockService{
+				config: apid.Config(),
+				log:    apid.Log(),
+				api:    ma,
+				data:   apid.Data(),
+				events: me,
+			}
+			testname := "test_" + strconv.Itoa(testCount)
+			ms.config.Set(configName, testname)
+			pd, err := initPlugin(ms)
+			Expect(err).Should(Succeed())
+			Expect(apidInfo.InstanceName).To(Equal(testname))
+			Expect(me.listenerMap[apid.SystemEventsSelector]).ToNot(BeNil())
+			Expect(ma.handleMap[tokenEndpoint]).ToNot(BeNil())
+			Expect(pd).Should(Equal(PluginData))
+			Expect(apidInfo.IsNewInstance).Should(BeTrue())
+		})
 
-			initConfigDefaults()
-			Expect(apidInfo.InstanceName).To(Equal("testhost"))
-		}, 3)
+		It("create managers for normal mode", func() {
+			listenerMan, apiMan, err := initManagers(false)
+			Expect(err).Should(Succeed())
+			Expect(listenerMan).ToNot(BeNil())
+			Expect(listenerMan.tokenMan).ToNot(BeNil())
+			snapMan, ok := listenerMan.snapMan.(*apidSnapshotManager)
+			Expect(ok).Should(BeTrue())
+			Expect(snapMan.tokenMan).ToNot(BeNil())
+			Expect(snapMan.dbMan).ToNot(BeNil())
+			changeMan, ok := listenerMan.changeMan.(*pollChangeManager)
+			Expect(ok).Should(BeTrue())
+			Expect(changeMan.tokenMan).ToNot(BeNil())
+			Expect(changeMan.dbMan).ToNot(BeNil())
+			Expect(changeMan.snapMan).ToNot(BeNil())
+			Expect(apiMan).ToNot(BeNil())
+			Expect(apiMan.tokenMan).ToNot(BeNil())
+		})
 
-		It("accept display name from config", func() {
-			config.Set(configName, "aa01")
-			initConfigDefaults()
-			var apidInfoLatest apidInstanceInfo
-			apidInfoLatest, _ = getApidInstanceInfo()
-			Expect(apidInfoLatest.InstanceName).To(Equal("aa01"))
-			Expect(apidInfoLatest.LastSnapshot).To(Equal(""))
-		}, 3)
+		It("create managers for diagnostic mode", func() {
+			config.Set(configDiagnosticMode, true)
+			listenerMan, apiMan, err := initManagers(true)
+			Expect(err).Should(Succeed())
+			Expect(listenerMan).ToNot(BeNil())
+			Expect(listenerMan.tokenMan).ToNot(BeNil())
+			snapMan, ok := listenerMan.snapMan.(*offlineSnapshotManager)
+			Expect(ok).Should(BeTrue())
+			Expect(snapMan.dbMan).ToNot(BeNil())
+			_, ok = listenerMan.changeMan.(*offlineChangeManager)
+			Expect(ok).Should(BeTrue())
+			Expect(apiMan).ToNot(BeNil())
+			Expect(apiMan.tokenMan).ToNot(BeNil())
+		})
 
 	})
-
-	It("should put apigeesync_apid_instance_id value in config", func() {
-		instanceID := config.GetString(configApidInstanceID)
-		Expect(instanceID).NotTo(BeEmpty())
-		Expect(instanceID).To(Equal(apidInfo.InstanceID))
-	})
 })
diff --git a/listener.go b/listener.go
index 94befd6..7f46b34 100644
--- a/listener.go
+++ b/listener.go
@@ -15,9 +15,8 @@
 package apidApigeeSync
 
 import (
-	"errors"
+	"encoding/json"
 	"github.com/apid/apid-core"
-	"github.com/apigee-labs/transicator/common"
 )
 
 const (
@@ -25,107 +24,86 @@
 	LISTENER_TABLE_DATA_SCOPE   = "edgex.data_scope"
 )
 
-func processSnapshot(snapshot *common.Snapshot) {
+type listenerManager struct {
+	changeMan     changeManager
+	snapMan       snapshotManager
+	tokenMan      tokenManager
+	isOfflineMode bool
+}
 
-	var prevDb string
-	if apidInfo.LastSnapshot != "" && apidInfo.LastSnapshot != snapshot.SnapshotInfo {
-		log.Debugf("Release snapshot for {%s}. Switching to version {%s}",
-			apidInfo.LastSnapshot, snapshot.SnapshotInfo)
-		prevDb = apidInfo.LastSnapshot
-	} else {
-		log.Debugf("Process snapshot for version {%s}",
-			snapshot.SnapshotInfo)
-	}
-	db, err := dataService.DBVersion(snapshot.SnapshotInfo)
-	if err != nil {
-		log.Panicf("Unable to access database: %v", err)
-	}
+func (l *listenerManager) init() {
+	/* This callback function will get called once all the plugins are
+	 * initialized (not just this plugin). This is needed because,
+	 * downloadSnapshots/changes etc have to begin to be processed only
+	 * after all the plugins are initialized
+	 */
+	eventService.ListenOnceFunc(apid.SystemEventsSelector, l.postInitPlugins)
+}
 
-	processSqliteSnapshot(db)
+// Plugins have all initialized, gather their info and start the ApigeeSync downloads
+func (l *listenerManager) postInitPlugins(event apid.Event) {
+	var plinfoDetails []pluginDetail
+	if pie, ok := event.(apid.PluginsInitializedEvent); ok {
+		/*
+		 * Store the plugin details in the heap. Needed during
+		 * Bearer token generation request.
+		 */
+		for _, plugin := range pie.Plugins {
+			name := plugin.Name
+			version := plugin.Version
+			if schemaVersion, ok := plugin.ExtraData["schemaVersion"].(string); ok {
+				inf := pluginDetail{
+					Name:          name,
+					SchemaVersion: schemaVersion}
+				plinfoDetails = append(plinfoDetails, inf)
+				log.Debugf("plugin %s is version %s, schemaVersion: %s", name, version, schemaVersion)
+			}
+		}
+		if plinfoDetails == nil {
+			log.Panic("No Plugins registered!")
+		}
 
-	//update apid instance info
-	apidInfo.LastSnapshot = snapshot.SnapshotInfo
-	err = updateApidInstanceInfo()
-	if err != nil {
-		log.Panicf("Unable to update instance info: %v", err)
-	}
+		pgInfo, err := json.Marshal(plinfoDetails)
+		if err != nil {
+			log.Panicf("Unable to marshal plugin data: %v", err)
+		}
+		apidPluginDetails = string(pgInfo[:])
 
-	setDB(db)
-	log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
+		log.Debug("start post plugin init")
 
-	// Releases the DB, when the Connection reference count reaches 0.
-	if prevDb != "" {
-		dataService.ReleaseDB(prevDb)
+		l.tokenMan.start()
+		go l.bootstrap(apidInfo.LastSnapshot)
+
+		log.Debug("Done post plugin init")
 	}
 }
 
-func processSqliteSnapshot(db apid.DB) {
-
-	var numApidClusters int
-	tx, err := db.Begin()
-	if err != nil {
-		log.Panicf("Unable to open DB txn: {%v}", err.Error())
-	}
-	defer tx.Rollback()
-	err = tx.QueryRow("SELECT COUNT(*) FROM edgex_apid_cluster").Scan(&numApidClusters)
-	if err != nil {
-		log.Panicf("Unable to read database: {%s}", err.Error())
+/*
+ *  Start from existing snapshot if possible
+ *  If an existing snapshot does not exist, use the apid scope to fetch
+ *  all data scopes, then get a snapshot for those data scopes
+ *
+ *  Then, poll for changes
+ */
+func (l *listenerManager) bootstrap(lastSnap string) {
+	if l.isOfflineMode && lastSnap == "" {
+		log.Panic("Diagnostic mode requires existent snapshot info in default DB.")
 	}
 
-	if numApidClusters != 1 {
-		log.Panic("Illegal state for apid_cluster. Must be a single row.")
-	}
-
-	_, err = tx.Exec("ALTER TABLE edgex_apid_cluster ADD COLUMN last_sequence text DEFAULT ''")
-	if err != nil {
-		if err.Error() == "duplicate column name: last_sequence" {
+	if lastSnap != "" {
+		if err := l.snapMan.startOnDataSnapshot(lastSnap); err == nil {
+			log.Infof("Started on local snapshot: %s", lastSnap)
+			l.changeMan.pollChangeWithBackoff()
 			return
 		} else {
-			log.Panicf("Unable to create last_sequence column on DB.  Error {%v}", err.Error())
-		}
-	}
-	if err = tx.Commit(); err != nil {
-		log.Errorf("Error when commit in processSqliteSnapshot: %v", err)
-	}
-}
-
-func processChangeList(changes *common.ChangeList) bool {
-
-	ok := false
-
-	tx, err := getDB().Begin()
-	if err != nil {
-		log.Panicf("Error processing ChangeList: %v", err)
-	}
-	defer tx.Rollback()
-
-	log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
-
-	for _, change := range changes.Changes {
-		if change.Table == LISTENER_TABLE_APID_CLUSTER {
-			log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
-		}
-		switch change.Operation {
-		case common.Insert:
-			ok = insert(change.Table, []common.Row{change.NewRow}, tx)
-		case common.Update:
-			if change.Table == LISTENER_TABLE_DATA_SCOPE {
-				log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
-			}
-			ok = update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx)
-		case common.Delete:
-			ok = _delete(change.Table, []common.Row{change.OldRow}, tx)
-		}
-		if !ok {
-			err = errors.New("Sql Operation error. Operation rollbacked")
-			log.Error("Sql Operation error. Operation rollbacked")
-			return false
+			log.Errorf("Failed to bootstrap on local snapshot: %v", err)
+			log.Warn("Will get new snapshots.")
 		}
 	}
 
-	if err = tx.Commit(); err != nil {
-		log.Errorf("Commit error in processChangeList: %v", err)
-		return false
+	l.snapMan.downloadBootSnapshot()
+	if err := l.snapMan.downloadDataSnapshot(); err != nil {
+		log.Panicf("Error downloading data snapshot: %v", err)
 	}
-	return true
+	l.changeMan.pollChangeWithBackoff()
 }
diff --git a/listener_test.go b/listener_test.go
index 54974eb..5e8f072 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -15,360 +15,79 @@
 package apidApigeeSync
 
 import (
+	"github.com/apid/apid-core"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-
-	"github.com/apid/apid-core"
-	"github.com/apigee-labs/transicator/common"
-	"os"
-	"reflect"
-	"sort"
+	"strconv"
 )
 
 var _ = Describe("listener", func() {
+	testCount := 0
+	var testListenerMan *listenerManager
+	var dummyChangeMan *dummyChangeManager
+	var dummySnapMan *dummySnapshotManager
+	var dummyTokenMan *dummyTokenManager
 
-	var createTestDb = func(sqlfile string, dbId string) common.Snapshot {
-		initDb(sqlfile, "./mockdb.sqlite3")
-		file, err := os.Open("./mockdb.sqlite3")
-		Expect(err).ShouldNot(HaveOccurred())
-
-		s := common.Snapshot{}
-		err = processSnapshotServerFileResponse(dbId, file, &s)
-		Expect(err).ShouldNot(HaveOccurred())
-		return s
-	}
-
-	var _ = BeforeEach(func() {
-		_initPlugin(apid.AllServices())
-	})
-
-	var _ = AfterEach(func() {
-		if wipeDBAferTest {
-			db, err := dataService.DB()
-			Expect(err).Should(Succeed())
-			tx, err := db.Begin()
-			Expect(err).Should(Succeed())
-			_, err = tx.Exec("DELETE FROM APID")
-			Expect(err).Should(Succeed())
-			err = tx.Commit()
-			Expect(err).Should(Succeed())
+	BeforeEach(func() {
+		testCount++
+		dummySnapMan = &dummySnapshotManager{
+			downloadCalledChan: make(chan bool, 1),
+			startCalledChan:    make(chan bool, 1),
 		}
-		wipeDBAferTest = true
+		dummyTokenMan = &dummyTokenManager{
+			invalidateChan: make(chan bool, 1),
+		}
+		dummyChangeMan = &dummyChangeManager{
+			pollChangeWithBackoffChan: make(chan bool, 1),
+		}
+		testListenerMan = &listenerManager{
+			changeMan: dummyChangeMan,
+			snapMan:   dummySnapMan,
+			tokenMan:  dummyTokenMan,
+		}
 	})
 
-	Context("ApigeeSync snapshot event", func() {
+	AfterEach(func() {
 
-		It("should fail if more than one apid_cluster rows", func() {
-			event := createTestDb("./sql/init_listener_test_duplicate_apids.sql", "test_snapshot_fail_multiple_clusters")
-			Expect(func() { processSnapshot(&event) }).To(Panic())
-		}, 3)
-
-		It("should fail if more than one apid_cluster rows", func() {
-			newScopes := []string{"foo"}
-			scopes := []string{"bar"}
-			Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
-			newScopes = []string{"foo", "bar"}
-			scopes = []string{"bar"}
-			Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
-			newScopes = []string{"foo"}
-			scopes = []string{"bar", "foo"}
-			Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
-			newScopes = []string{"foo", "bar"}
-			scopes = []string{"bar", "foo"}
-			Expect(scopeChanged(newScopes, scopes)).To(BeNil())
-
-		}, 3)
-
-		It("should process a valid Snapshot", func() {
-
-			event := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_snapshot_valid")
-
-			processSnapshot(&event)
-
-			info, err := getApidInstanceInfo()
-			Expect(err).NotTo(HaveOccurred())
-
-			Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo))
-
-			db := getDB()
-
-			expectedDB, err := dataService.DBVersion(event.SnapshotInfo)
-			Expect(err).NotTo(HaveOccurred())
-
-			Expect(db == expectedDB).Should(BeTrue())
-
-			// apid Cluster
-			var dcs []dataApidCluster
-
-			rows, err := db.Query(`
-			SELECT id, name, description, umbrella_org_app_name,
-				created, created_by, updated, updated_by
-			FROM EDGEX_APID_CLUSTER`)
-			Expect(err).NotTo(HaveOccurred())
-			defer rows.Close()
-
-			c := dataApidCluster{}
-			for rows.Next() {
-				rows.Scan(&c.ID, &c.Name, &c.Description, &c.OrgAppName,
-					&c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy)
-				dcs = append(dcs, c)
-			}
-
-			Expect(len(dcs)).To(Equal(1))
-			dc := dcs[0]
-
-			Expect(dc.ID).To(Equal("i"))
-			Expect(dc.Name).To(Equal("n"))
-			Expect(dc.Description).To(Equal("d"))
-			Expect(dc.OrgAppName).To(Equal("o"))
-			Expect(dc.Created).To(Equal("c"))
-			Expect(dc.CreatedBy).To(Equal("c"))
-			Expect(dc.Updated).To(Equal("u"))
-			Expect(dc.UpdatedBy).To(Equal("u"))
-
-			// Data Scope
-			var dds []dataDataScope
-
-			rows, err = db.Query(`
-			SELECT id, apid_cluster_id, scope, org,
-				env, created, created_by, updated,
-				updated_by
-			FROM EDGEX_DATA_SCOPE`)
-			Expect(err).NotTo(HaveOccurred())
-			defer rows.Close()
-
-			d := dataDataScope{}
-			for rows.Next() {
-				rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
-					&d.Env, &d.Created, &d.CreatedBy, &d.Updated,
-					&d.UpdatedBy)
-				dds = append(dds, d)
-			}
-
-			Expect(len(dds)).To(Equal(3))
-			ds := dds[0]
-
-			Expect(ds.ID).To(Equal("i"))
-			Expect(ds.Org).To(Equal("o"))
-			Expect(ds.Env).To(Equal("e1"))
-			Expect(ds.Scope).To(Equal("s1"))
-			Expect(ds.Created).To(Equal("c"))
-			Expect(ds.CreatedBy).To(Equal("c"))
-			Expect(ds.Updated).To(Equal("u"))
-			Expect(ds.UpdatedBy).To(Equal("u"))
-
-			ds = dds[1]
-			Expect(ds.Env).To(Equal("e2"))
-			Expect(ds.Scope).To(Equal("s1"))
-			ds = dds[2]
-			Expect(ds.Env).To(Equal("e3"))
-			Expect(ds.Scope).To(Equal("s2"))
-
-			scopes := findScopesForId("a")
-			Expect(len(scopes)).To(Equal(6))
-			expectedScopes := []string{"s1", "s2", "org_scope_1", "env_scope_1", "env_scope_2", "env_scope_3"}
-			sort.Strings(scopes)
-			sort.Strings(expectedScopes)
-			Expect(reflect.DeepEqual(scopes, expectedScopes)).To(BeTrue())
-		}, 3)
 	})
 
-	Context("ApigeeSync change event", func() {
-
-		Context(LISTENER_TABLE_APID_CLUSTER, func() {
-
-			It("insert event should panic", func() {
-				ssEvent := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_changes_insert_panic")
-				processSnapshot(&ssEvent)
-
-				//save the last snapshot, so we can restore it at the end of this context
-
-				csEvent := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Insert,
-							Table:     LISTENER_TABLE_APID_CLUSTER,
-						},
+	It("postInitPlugins, start cleanly", func() {
+		testEvent := apid.EventSelector("test event" + strconv.Itoa(testCount))
+		eventService.ListenOnceFunc(testEvent, testListenerMan.postInitPlugins)
+		eventService.Emit(testEvent, apid.PluginsInitializedEvent{
+			Description: "test",
+			Plugins: []apid.PluginData{
+				{
+					Name:    "name",
+					Version: "0.0.1",
+					ExtraData: map[string]interface{}{
+						"schemaVersion": "0.0.1",
 					},
-				}
-
-				Expect(func() { processChangeList(&csEvent) }).To(Panic())
-			}, 3)
-
-			It("update event should panic", func() {
-				ssEvent := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_changes_update_panic")
-				processSnapshot(&ssEvent)
-
-				event := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Update,
-							Table:     LISTENER_TABLE_APID_CLUSTER,
-						},
-					},
-				}
-
-				Expect(func() { processChangeList(&event) }).To(Panic())
-				//restore the last snapshot
-			}, 3)
-
+				},
+			},
 		})
-
-		Context(LISTENER_TABLE_DATA_SCOPE, func() {
-
-			It("insert event should add", func() {
-				ssEvent := createTestDb("./sql/init_listener_test_no_datascopes.sql", "test_changes_insert")
-				processSnapshot(&ssEvent)
-
-				event := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Insert,
-							Table:     LISTENER_TABLE_DATA_SCOPE,
-							NewRow: common.Row{
-								"id":               &common.ColumnVal{Value: "i"},
-								"apid_cluster_id":  &common.ColumnVal{Value: "a"},
-								"scope":            &common.ColumnVal{Value: "s1"},
-								"org":              &common.ColumnVal{Value: "o"},
-								"env":              &common.ColumnVal{Value: "e"},
-								"created":          &common.ColumnVal{Value: "c"},
-								"created_by":       &common.ColumnVal{Value: "c"},
-								"updated":          &common.ColumnVal{Value: "u"},
-								"updated_by":       &common.ColumnVal{Value: "u"},
-								"_change_selector": &common.ColumnVal{Value: "cs"},
-							},
-						},
-						{
-							Operation: common.Insert,
-							Table:     LISTENER_TABLE_DATA_SCOPE,
-							NewRow: common.Row{
-								"id":               &common.ColumnVal{Value: "j"},
-								"apid_cluster_id":  &common.ColumnVal{Value: "a"},
-								"scope":            &common.ColumnVal{Value: "s2"},
-								"org":              &common.ColumnVal{Value: "o"},
-								"env":              &common.ColumnVal{Value: "e"},
-								"created":          &common.ColumnVal{Value: "c"},
-								"created_by":       &common.ColumnVal{Value: "c"},
-								"updated":          &common.ColumnVal{Value: "u"},
-								"updated_by":       &common.ColumnVal{Value: "u"},
-								"_change_selector": &common.ColumnVal{Value: "cs"},
-							},
-						},
-					},
-				}
-
-				processChangeList(&event)
-
-				var dds []dataDataScope
-
-				rows, err := getDB().Query(`
-				SELECT id, apid_cluster_id, scope, org,
-					env, created, created_by, updated,
-					updated_by
-				FROM EDGEX_DATA_SCOPE`)
-				Expect(err).NotTo(HaveOccurred())
-				defer rows.Close()
-
-				d := dataDataScope{}
-				for rows.Next() {
-					rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
-						&d.Env, &d.Created, &d.CreatedBy, &d.Updated,
-						&d.UpdatedBy)
-					dds = append(dds, d)
-				}
-
-				//three already existing
-				Expect(len(dds)).To(Equal(2))
-				ds := dds[0]
-
-				Expect(ds.ID).To(Equal("i"))
-				Expect(ds.Org).To(Equal("o"))
-				Expect(ds.Env).To(Equal("e"))
-				Expect(ds.Scope).To(Equal("s1"))
-				Expect(ds.Created).To(Equal("c"))
-				Expect(ds.CreatedBy).To(Equal("c"))
-				Expect(ds.Updated).To(Equal("u"))
-				Expect(ds.UpdatedBy).To(Equal("u"))
-
-				ds = dds[1]
-				Expect(ds.Scope).To(Equal("s2"))
-
-				scopes := findScopesForId("a")
-				Expect(len(scopes)).To(Equal(2))
-				Expect(scopes[0]).To(Equal("s1"))
-				Expect(scopes[1]).To(Equal("s2"))
-
-			}, 3)
-
-			It("delete event should delete", func() {
-				ssEvent := createTestDb("./sql/init_listener_test_no_datascopes.sql", "test_changes_delete")
-				processSnapshot(&ssEvent)
-				insert := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Insert,
-							Table:     LISTENER_TABLE_DATA_SCOPE,
-							NewRow: common.Row{
-								"id":               &common.ColumnVal{Value: "i"},
-								"apid_cluster_id":  &common.ColumnVal{Value: "a"},
-								"scope":            &common.ColumnVal{Value: "s"},
-								"org":              &common.ColumnVal{Value: "o"},
-								"env":              &common.ColumnVal{Value: "e"},
-								"created":          &common.ColumnVal{Value: "c"},
-								"created_by":       &common.ColumnVal{Value: "c"},
-								"updated":          &common.ColumnVal{Value: "u"},
-								"updated_by":       &common.ColumnVal{Value: "u"},
-								"_change_selector": &common.ColumnVal{Value: "cs"},
-							},
-						},
-					},
-				}
-
-				processChangeList(&insert)
-
-				delete := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Delete,
-							Table:     LISTENER_TABLE_DATA_SCOPE,
-							OldRow:    insert.Changes[0].NewRow,
-						},
-					},
-				}
-
-				processChangeList(&delete)
-
-				var nRows int
-				err := getDB().QueryRow("SELECT count(id) FROM EDGEX_DATA_SCOPE").Scan(&nRows)
-				Expect(err).NotTo(HaveOccurred())
-
-				Expect(0).To(Equal(nRows))
-			}, 3)
-
-			It("update event should panic for data scopes table", func() {
-				ssEvent := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_update_panic")
-				processSnapshot(&ssEvent)
-
-				event := common.ChangeList{
-					LastSequence: "test",
-					Changes: []common.Change{
-						{
-							Operation: common.Update,
-							Table:     LISTENER_TABLE_DATA_SCOPE,
-						},
-					},
-				}
-
-				Expect(func() { processChangeList(&event) }).To(Panic())
-				//restore the last snapshot
-			}, 3)
-
-			//TODO add tests for update/insert/delete cluster
-		})
+		Expect(<-dummySnapMan.downloadCalledChan).Should(BeFalse())
+		Expect(<-dummyChangeMan.pollChangeWithBackoffChan).Should(BeTrue())
 	})
+
+	It("postInitPlugins, start from local db", func() {
+		apidInfo.LastSnapshot = "test_snapshot"
+		testEvent := apid.EventSelector("test event" + strconv.Itoa(testCount))
+		eventService.ListenOnceFunc(testEvent, testListenerMan.postInitPlugins)
+		eventService.Emit(testEvent, apid.PluginsInitializedEvent{
+			Description: "test",
+			Plugins: []apid.PluginData{
+				{
+					Name:    "name",
+					Version: "0.0.1",
+					ExtraData: map[string]interface{}{
+						"schemaVersion": "0.0.1",
+					},
+				},
+			},
+		})
+		Expect(<-dummySnapMan.startCalledChan).Should(BeTrue())
+		Expect(<-dummyChangeMan.pollChangeWithBackoffChan).Should(BeTrue())
+	})
+
 })
diff --git a/managerInterfaces.go b/managerInterfaces.go
index a93e950..978aede 100644
--- a/managerInterfaces.go
+++ b/managerInterfaces.go
@@ -15,31 +15,38 @@
 package apidApigeeSync
 
 import (
+	"github.com/apid/apid-core"
 	"github.com/apigee-labs/transicator/common"
-	"net/url"
 )
 
 type tokenManager interface {
 	getBearerToken() string
-	invalidateToken() error
-	getToken() *OauthToken
+	invalidateToken()
 	close()
-	getRetrieveNewTokenClosure(*url.URL) func(chan bool) error
 	start()
 	getTokenReadyChannel() <-chan bool
 }
 
-type snapShotManager interface {
+type snapshotManager interface {
 	close() <-chan bool
 	downloadBootSnapshot()
-	storeBootSnapshot(snapshot *common.Snapshot)
-	downloadDataSnapshot()
-	storeDataSnapshot(snapshot *common.Snapshot)
-	downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error
-	startOnLocalSnapshot(snapshot string) *common.Snapshot
+	downloadDataSnapshot() error
+	startOnDataSnapshot(snapshot string) error
 }
 
 type changeManager interface {
 	close() <-chan bool
 	pollChangeWithBackoff()
 }
+
+type DbManager interface {
+	initDB() error
+	setDB(db apid.DB)
+	getLastSequence() (lastSequence string)
+	findScopesForId(configId string) (scopes []string, err error)
+	updateLastSequence(lastSequence string) error
+	getApidInstanceInfo() (info apidInstanceInfo, err error)
+	processChangeList(changes *common.ChangeList) error
+	processSnapshot(snapshot *common.Snapshot, isDataSnapshot bool) error
+	getKnowTables() map[string]bool
+}
diff --git a/mock_server.go b/mock_server_test.go
similarity index 94%
rename from mock_server.go
rename to mock_server_test.go
index f39adb7..e61b83e 100644
--- a/mock_server.go
+++ b/mock_server_test.go
@@ -90,14 +90,14 @@
 	changeChannel   chan []byte
 	sequenceID      *int64
 	maxDevID        *int64
-	deployIDMutex   sync.RWMutex
+	deployIDMutex   *sync.RWMutex
 	minDeploymentID *int64
 	maxDeploymentID *int64
 	newSnap         *int32
 	authFail        *int32
 }
 
-func (m *MockServer) forceAuthFail() {
+func (m *MockServer) forceAuthFailOnce() {
 	atomic.StoreInt32(m.authFail, 1)
 }
 
@@ -118,11 +118,13 @@
 }
 
 func (m *MockServer) lastSequenceID() string {
-	return strconv.FormatInt(atomic.LoadInt64(m.sequenceID), 10)
+	num := strconv.FormatInt(atomic.LoadInt64(m.sequenceID), 10)
+	return num + "." + num + "." + num
 }
 
 func (m *MockServer) nextSequenceID() string {
-	return strconv.FormatInt(atomic.AddInt64(m.sequenceID, 1), 10)
+	num := strconv.FormatInt(atomic.AddInt64(m.sequenceID, 1), 10)
+	return num + "." + num + "." + num
 }
 
 func (m *MockServer) nextDeveloperID() string {
@@ -180,7 +182,7 @@
 	m.newSnap = new(int32)
 	m.authFail = new(int32)
 	*m.authFail = 0
-
+	m.deployIDMutex = &sync.RWMutex{}
 	initDb("./sql/init_mock_db.sql", "./mockdb.sqlite3")
 	initDb("./sql/init_mock_boot_db.sql", "./mockdb_boot.sqlite3")
 
@@ -266,8 +268,11 @@
 	scopes := q["scope"]
 
 	Expect(scopes).To(ContainElement(m.params.ClusterID))
-
-	w.Header().Set("Transicator-Snapshot-TXID", util.GenerateUUID())
+	if m.params.Scope != "" {
+		Expect(scopes).To(ContainElement(m.params.Scope))
+	}
+	m.snapshotID = util.GenerateUUID()
+	w.Header().Set(headerSnapshotNumber, m.snapshotID)
 
 	if len(scopes) == 1 {
 		//send bootstrap db
@@ -285,7 +290,7 @@
 func (m *MockServer) sendChanges(w http.ResponseWriter, req *http.Request) {
 	defer GinkgoRecover()
 
-	val := atomic.SwapInt32(m.newSnap, 0)
+	val := atomic.LoadInt32(m.newSnap)
 	if val > 0 {
 		log.Debug("MockServer: force new snapshot")
 		w.WriteHeader(http.StatusBadRequest)
@@ -311,7 +316,9 @@
 	//Expect(q.Get("snapshot")).To(Equal(m.snapshotID))
 
 	Expect(scopes).To(ContainElement(m.params.ClusterID))
-	//Expect(scopes).To(ContainElement(m.params.Scope))
+	if m.params.Scope != "" {
+		Expect(scopes).To(ContainElement(m.params.Scope))
+	}
 
 	// todo: the following is just legacy for the existing test in apigeeSync_suite_test
 	developer := m.createDeveloperWithProductAndApp()
@@ -343,6 +350,7 @@
 
 		// force failing auth check
 		if atomic.LoadInt32(m.authFail) == 1 {
+			atomic.StoreInt32(m.authFail, 0)
 			w.WriteHeader(http.StatusUnauthorized)
 			w.Write([]byte(fmt.Sprintf("Force fail: bad auth token. ")))
 			return
@@ -356,7 +364,7 @@
 
 		// check auth header
 		auth := req.Header.Get("Authorization")
-		expectedAuth := fmt.Sprintf("Bearer %s", m.oauthToken)
+		expectedAuth := m.getBearerToken()
 		if auth != expectedAuth {
 			w.WriteHeader(http.StatusUnauthorized)
 			w.Write([]byte(fmt.Sprintf("Bad auth token. Is: %s, should be: %s", auth, expectedAuth)))
@@ -366,6 +374,10 @@
 	}
 }
 
+func (m *MockServer) getBearerToken() string {
+	return fmt.Sprintf("Bearer %s", m.oauthToken)
+}
+
 // make a handler unreliable
 func (m *MockServer) unreliable(target http.HandlerFunc) http.HandlerFunc {
 	if m.params.ReliableAPI {
diff --git a/pluginData.go b/pluginData.go
index c565c02..10b2dd0 100644
--- a/pluginData.go
+++ b/pluginData.go
@@ -16,7 +16,7 @@
 
 import "github.com/apid/apid-core"
 
-var pluginData = apid.PluginData{
+var PluginData = apid.PluginData{
 	Name:    "apidApigeeSync",
 	Version: "0.0.4",
 	ExtraData: map[string]interface{}{
diff --git a/snapshot.go b/snapshot.go
index df4d63f..023314d 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -15,12 +15,12 @@
 package apidApigeeSync
 
 import (
-	"github.com/apid/apid-core"
 	"github.com/apid/apid-core/data"
 	"github.com/apigee-labs/transicator/common"
 	"net/http"
 	"os"
 
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net/url"
@@ -29,7 +29,14 @@
 	"time"
 )
 
-type simpleSnapShotManager struct {
+const bootstrapSnapshotName = "bootstrap"
+const lengthSqliteFileName = 7 // len("/sqlite")
+const (
+	headerSnapshotNumber = "Transicator-Snapshot-TXID"
+)
+
+type apidSnapshotManager struct {
+	*offlineSnapshotManager
 	// to send quit signal to the downloading thread
 	quitChan chan bool
 	// to mark the graceful close of snapshotManager
@@ -38,32 +45,40 @@
 	isClosed *int32
 	// make sure close() returns immediately if there's no downloading/processing snapshot
 	isDownloading *int32
+	tokenMan      tokenManager
+	dbMan         DbManager
+	client        *http.Client
 }
 
-func createSnapShotManager() *simpleSnapShotManager {
+func createSnapShotManager(dbMan DbManager, tokenMan tokenManager, client *http.Client) *apidSnapshotManager {
 	isClosedInt := int32(0)
 	isDownloadingInt := int32(0)
-	return &simpleSnapShotManager{
+	return &apidSnapshotManager{
+		offlineSnapshotManager: &offlineSnapshotManager{
+			dbMan: dbMan,
+		},
 		quitChan:      make(chan bool, 1),
 		finishChan:    make(chan bool, 1),
 		isClosed:      &isClosedInt,
 		isDownloading: &isDownloadingInt,
+		dbMan:         dbMan,
+		tokenMan:      tokenMan,
+		client:        client,
 	}
 }
 
 /*
- * thread-safe close of snapShotManager
+ * thread-safe close of snapshotManager
  * It marks status as closed immediately, and quits backoff downloading
  * use <- close() for blocking close
  * should only be called by pollChangeManager, because pollChangeManager is dependent on it
  */
-func (s *simpleSnapShotManager) close() <-chan bool {
+func (s *apidSnapshotManager) close() <-chan bool {
 	//has been closed before
 	if atomic.SwapInt32(s.isClosed, 1) == int32(1) {
-		log.Error("snapShotManager: close() called on a closed snapShotManager!")
+		log.Warn("snapshotManager: close() called on a closed snapshotManager!")
 		go func() {
 			s.finishChan <- false
-			log.Debug("change manager closed")
 		}()
 		return s.finishChan
 	}
@@ -77,163 +92,52 @@
 }
 
 // retrieve boot information: apid_config and apid_config_scope
-func (s *simpleSnapShotManager) downloadBootSnapshot() {
+func (s *apidSnapshotManager) downloadBootSnapshot() {
 	if atomic.SwapInt32(s.isDownloading, 1) == int32(1) {
 		log.Panic("downloadBootSnapshot: only 1 thread can download snapshot at the same time!")
 	}
 	defer atomic.StoreInt32(s.isDownloading, int32(0))
 
-	// has been closed
-	if atomic.LoadInt32(s.isClosed) == int32(1) {
-		log.Warn("snapShotManager: downloadBootSnapshot called on closed snapShotManager")
-		return
-	}
-
 	log.Debug("download Snapshot for boot data")
 
 	scopes := []string{apidInfo.ClusterID}
 	snapshot := &common.Snapshot{}
 
-	err := s.downloadSnapshot(true, scopes, snapshot)
-	if err != nil {
-		// this may happen during shutdown
-		if _, ok := err.(quitSignalError); ok {
-			log.Warn("downloadBootSnapshot failed due to shutdown: " + err.Error())
-		}
-		return
-	}
-
-	// has been closed
-	if atomic.LoadInt32(s.isClosed) == int32(1) {
-		log.Error("snapShotManager: processSnapshot called on closed snapShotManager")
-		return
-	}
+	s.downloadSnapshot(true, scopes, snapshot)
 
 	// note that for boot snapshot case, we don't need to inform plugins as they'll get the data snapshot
 	s.storeBootSnapshot(snapshot)
 }
 
-func (s *simpleSnapShotManager) storeBootSnapshot(snapshot *common.Snapshot) {
-	processSnapshot(snapshot)
+func (s *apidSnapshotManager) storeBootSnapshot(snapshot *common.Snapshot) {
+	if err := s.dbMan.processSnapshot(snapshot, false); err != nil {
+		log.Panic(err)
+	}
 }
 
 // use the scope IDs from the boot snapshot to get all the data associated with the scopes
-func (s *simpleSnapShotManager) downloadDataSnapshot() {
+func (s *apidSnapshotManager) downloadDataSnapshot() error {
 	if atomic.SwapInt32(s.isDownloading, 1) == int32(1) {
 		log.Panic("downloadDataSnapshot: only 1 thread can download snapshot at the same time!")
 	}
 	defer atomic.StoreInt32(s.isDownloading, int32(0))
 
-	// has been closed
-	if atomic.LoadInt32(s.isClosed) == int32(1) {
-		log.Warn("snapShotManager: downloadDataSnapshot called on closed snapShotManager")
-		return
-	}
-
 	log.Debug("download Snapshot for data scopes")
 
-	scopes := findScopesForId(apidInfo.ClusterID)
+	scopes, err := s.dbMan.findScopesForId(apidInfo.ClusterID)
+	if err != nil {
+		return err
+	}
 	scopes = append(scopes, apidInfo.ClusterID)
 	snapshot := &common.Snapshot{}
-	err := s.downloadSnapshot(false, scopes, snapshot)
-	if err != nil {
-		// this may happen during shutdown
-		if _, ok := err.(quitSignalError); ok {
-			log.Warn("downloadDataSnapshot failed due to shutdown: " + err.Error())
-		}
-		return
-	}
-	s.storeDataSnapshot(snapshot)
-}
-
-func (s *simpleSnapShotManager) storeDataSnapshot(snapshot *common.Snapshot) {
-	knownTables = extractTablesFromSnapshot(snapshot)
-
-	_, err := dataService.DBVersion(snapshot.SnapshotInfo)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
-
-	processSnapshot(snapshot)
-	log.Info("Emitting Snapshot to plugins")
-
-	select {
-	case <-time.After(pluginTimeout):
-		log.Panic("Timeout. Plugins failed to respond to snapshot.")
-	case <-events.Emit(ApigeeSyncEventSelector, snapshot):
-		// the new snapshot has been processed
-		// if close() happen after persistKnownTablesToDB(), will not interrupt snapshot processing to maintain consistency
-	}
-
-}
-
-func extractTablesFromSnapshot(snapshot *common.Snapshot) (tables map[string]bool) {
-
-	tables = make(map[string]bool)
-
-	log.Debug("Extracting table names from snapshot")
-	//if this panic ever fires, it's a bug
-	db, err := dataService.DBVersion(snapshot.SnapshotInfo)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
-	return extractTablesFromDB(db)
-}
-
-func extractTablesFromDB(db apid.DB) (tables map[string]bool) {
-
-	tables = make(map[string]bool)
-
-	log.Debug("Extracting table names from existing DB")
-	rows, err := db.Query("SELECT DISTINCT tableName FROM _transicator_tables;")
-	defer rows.Close()
-
-	if err != nil {
-		log.Panicf("Error reading current set of tables: %v", err)
-	}
-
-	for rows.Next() {
-		var table string
-		if err := rows.Scan(&table); err != nil {
-			log.Panicf("Error reading current set of tables: %v", err)
-		}
-		log.Debugf("Table %s found in existing db", table)
-
-		tables[table] = true
-	}
-	return tables
-}
-
-// Skip Downloading snapshot if there is already a snapshot available from previous run
-func (s *simpleSnapShotManager) startOnLocalSnapshot(snapshotName string) *common.Snapshot {
-	log.Infof("Starting on local snapshot: %s", snapshotName)
-
-	// ensure DB version will be accessible on behalf of dependant plugins
-	db, err := dataService.DBVersion(snapshotName)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
-
-	knownTables = extractTablesFromDB(db)
-	snapshot := &common.Snapshot{
-		SnapshotInfo: snapshotName,
-	}
-	processSnapshot(snapshot)
-
-	// allow plugins (including this one) to start immediately on existing database
-	// Note: this MUST have no tables as that is used as an indicator
-	return snapshot
+	s.downloadSnapshot(false, scopes, snapshot)
+	return s.startOnDataSnapshot(snapshot.SnapshotInfo)
 }
 
 // a blocking method
 // will keep retrying with backoff until success
 
-func (s *simpleSnapShotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error {
-	// if closed
-	if atomic.LoadInt32(s.isClosed) == int32(1) {
-		log.Warn("Trying to download snapshot with a closed snapShotManager")
-		return quitSignalError{}
-	}
+func (s *apidSnapshotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) {
 
 	log.Debug("downloadSnapshot")
 
@@ -254,12 +158,11 @@
 
 	//pollWithBackoff only accepts function that accept a single quit channel
 	//to accommodate functions which need more parameters, wrap them in closures
-	attemptDownload := getAttemptDownloadClosure(isBoot, snapshot, uri)
+	attemptDownload := s.getAttemptDownloadClosure(isBoot, snapshot, uri)
 	pollWithBackoff(s.quitChan, attemptDownload, handleSnapshotServerError)
-	return nil
 }
 
-func getAttemptDownloadClosure(isBoot bool, snapshot *common.Snapshot, uri string) func(chan bool) error {
+func (s *apidSnapshotManager) getAttemptDownloadClosure(isBoot bool, snapshot *common.Snapshot, uri string) func(chan bool) error {
 	return func(_ chan bool) error {
 
 		var tid string
@@ -268,19 +171,16 @@
 			// should never happen, but if it does, it's unrecoverable anyway
 			log.Panicf("Snapshotserver comm error: %v", err)
 		}
-		addHeaders(req)
-
-		var processSnapshotResponse func(string, io.Reader, *common.Snapshot) error
+		addHeaders(req, s.tokenMan.getBearerToken())
 
 		if config.GetString(configSnapshotProtocol) != "sqlite" {
 			log.Panic("Only currently supported snashot protocol is sqlite")
 
 		}
 		req.Header.Set("Accept", "application/transicator+sqlite")
-		processSnapshotResponse = processSnapshotServerFileResponse
 
 		// Issue the request to the snapshot server
-		r, err := httpclient.Do(req)
+		r, err := s.client.Do(req)
 		if err != nil {
 			log.Errorf("Snapshotserver comm error: %v", err)
 			return err
@@ -288,22 +188,28 @@
 
 		defer r.Body.Close()
 
-		if r.StatusCode != 200 {
+		switch r.StatusCode {
+		case http.StatusOK:
+			break
+		case http.StatusUnauthorized:
+			s.tokenMan.invalidateToken()
+			fallthrough
+		default:
 			body, _ := ioutil.ReadAll(r.Body)
 			log.Errorf("Snapshot server conn failed with resp code %d, body: %s", r.StatusCode, string(body))
-			return expected200Error{}
+			return expected200Error
 		}
 
 		// Bootstrap scope is a special case, that can occur only once. The tid is
 		// hardcoded to "bootstrap" to ensure there can be no clash of tid between
 		// bootstrap and subsequent data scopes.
 		if isBoot {
-			tid = "bootstrap"
+			tid = bootstrapSnapshotName
 		} else {
-			tid = r.Header.Get("Transicator-Snapshot-TXID")
+			tid = r.Header.Get(headerSnapshotNumber)
 		}
 		// Decode the Snapshot server response
-		err = processSnapshotResponse(tid, r.Body, snapshot)
+		err = processSnapshotServerFileResponse(tid, r.Body, snapshot)
 		if err != nil {
 			log.Errorf("Snapshot server response Data not parsable: %v", err)
 			return err
@@ -315,11 +221,22 @@
 
 func processSnapshotServerFileResponse(dbId string, body io.Reader, snapshot *common.Snapshot) error {
 	dbPath := data.DBPath("common/" + dbId)
+	dbDir := dbPath[0 : len(dbPath)-lengthSqliteFileName]
 	log.Infof("Attempting to stream the sqlite snapshot to %s", dbPath)
 
+	// if other bootstrap snapshot exists, delete the old file
+	if dbId == bootstrapSnapshotName {
+		if _, err := os.Stat(dbDir); !os.IsNotExist(err) {
+			if err = os.RemoveAll(dbDir); err != nil {
+				log.Errorf("Failed to delete old bootstrap snapshot; %v", err)
+				return err
+			}
+		}
+	}
+
 	//this path includes the sqlite3 file name.  why does mkdir all stop at parent??
 	log.Infof("Creating directory with mkdirall %s", dbPath)
-	err := os.MkdirAll(dbPath[0:len(dbPath)-7], 0700)
+	err := os.MkdirAll(dbDir, 0700)
 	if err != nil {
 		log.Errorf("Error creating db path %s", err)
 	}
@@ -343,10 +260,11 @@
 }
 
 func handleSnapshotServerError(err error) {
-	log.Debugf("Error connecting to snapshot server: %v", err)
+	log.Errorf("Error connecting to snapshot server: %v", err)
 }
 
 type offlineSnapshotManager struct {
+	dbMan DbManager
 }
 
 func (o *offlineSnapshotManager) close() <-chan bool {
@@ -355,33 +273,28 @@
 	return c
 }
 
-func (o *offlineSnapshotManager) downloadBootSnapshot() {}
-
-func (o *offlineSnapshotManager) storeBootSnapshot(snapshot *common.Snapshot) {}
-
-func (o *offlineSnapshotManager) downloadDataSnapshot() {}
-
-func (o *offlineSnapshotManager) storeDataSnapshot(snapshot *common.Snapshot) {}
-
-func (o *offlineSnapshotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error {
-	return nil
+func (o *offlineSnapshotManager) downloadBootSnapshot() {
+	log.Panic("downloadBootSnapshot called for offlineSnapshotManager")
 }
-func (o *offlineSnapshotManager) startOnLocalSnapshot(snapshotName string) *common.Snapshot {
-	log.Infof("Starting on local snapshot: %s", snapshotName)
 
-	// ensure DB version will be accessible on behalf of dependant plugins
-	db, err := dataService.DBVersion(snapshotName)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
+func (o *offlineSnapshotManager) downloadDataSnapshot() error {
+	return fmt.Errorf("downloadDataSnapshot called for offlineSnapshotManager")
+}
 
-	knownTables = extractTablesFromDB(db)
+func (o *offlineSnapshotManager) startOnDataSnapshot(snapshotName string) error {
+	log.Infof("Processing snapshot: %s", snapshotName)
 	snapshot := &common.Snapshot{
 		SnapshotInfo: snapshotName,
 	}
-	processSnapshot(snapshot)
-
-	// allow plugins (including this one) to start immediately on existing database
-	// Note: this MUST have no tables as that is used as an indicator
-	return snapshot
+	if err := o.dbMan.processSnapshot(snapshot, true); err != nil {
+		return err
+	}
+	log.Info("Emitting Snapshot to plugins")
+	select {
+	case <-time.After(pluginTimeout):
+		return fmt.Errorf("timeout, plugins failed to respond to snapshot")
+	case <-eventService.Emit(ApigeeSyncEventSelector, snapshot):
+		// the new snapshot has been processed
+	}
+	return nil
 }
diff --git a/snapshot_test.go b/snapshot_test.go
new file mode 100644
index 0000000..8d99563
--- /dev/null
+++ b/snapshot_test.go
@@ -0,0 +1,146 @@
+// Copyright 2017 Google Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apidApigeeSync
+
+import (
+	"github.com/apid/apid-core"
+	"github.com/apid/apid-core/api"
+	"github.com/apigee-labs/transicator/common"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"time"
+)
+
+var _ = Describe("Snapshot Manager", func() {
+	testCount := 0
+	var dummyDbMan *dummyDbManager
+	BeforeEach(func() {
+		testCount++
+		dummyDbMan = &dummyDbManager{}
+	})
+
+	Context("offlineSnapshotManager", func() {
+		var testSnapMan *offlineSnapshotManager
+		BeforeEach(func() {
+			testSnapMan = &offlineSnapshotManager{
+				dbMan: dummyDbMan,
+			}
+		})
+		AfterEach(func() {
+			<-testSnapMan.close()
+		})
+
+		It("should have error if download called", func() {
+			Expect(testSnapMan.downloadDataSnapshot()).ToNot(Succeed())
+			Expect(func() { testSnapMan.downloadBootSnapshot() }).To(Panic())
+		})
+
+		It("startOnDataSnapshot should emit events", func() {
+			called := false
+			eventService.ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+				if _, ok := event.(*common.Snapshot); ok {
+					called = true
+				}
+			})
+			snapshotId := "test_snapshot_" + strconv.Itoa(testCount)
+			Expect(testSnapMan.startOnDataSnapshot(snapshotId)).Should(Succeed())
+			Expect(dummyDbMan.snapshot.SnapshotInfo).Should(Equal(snapshotId))
+			Expect(called).Should(BeTrue())
+		})
+	})
+
+	Context("apidSnapshotManager", func() {
+		var testSnapMan *apidSnapshotManager
+		var dummyTokenMan *dummyTokenManager
+		var testServer *httptest.Server
+		var testRouter apid.Router
+		var testMock *MockServer
+		BeforeEach(func() {
+			dummyTokenMan = &dummyTokenManager{
+				invalidateChan: make(chan bool, 1),
+			}
+			client := &http.Client{}
+			testSnapMan = createSnapShotManager(dummyDbMan, dummyTokenMan, client)
+
+			// create a new API service to have a new router for testing
+			testRouter = api.CreateService().Router()
+			testServer = httptest.NewServer(testRouter)
+			// set up mock server
+			mockParms := MockParms{
+				ReliableAPI:  true,
+				ClusterID:    config.GetString(configApidClusterId),
+				TokenKey:     config.GetString(configConsumerKey),
+				TokenSecret:  config.GetString(configConsumerSecret),
+				Scope:        "",
+				Organization: "att",
+				Environment:  "prod",
+			}
+			apidInfo.ClusterID = expectedClusterId
+			apidInfo.InstanceID = expectedInstanceId
+			testMock = Mock(mockParms, testRouter)
+			config.Set(configProxyServerBaseURI, testServer.URL)
+			config.Set(configSnapServerBaseURI, testServer.URL)
+			config.Set(configChangeServerBaseURI, testServer.URL)
+			config.Set(configPollInterval, 1*time.Millisecond)
+
+			initialBackoffInterval = time.Millisecond
+			testMock.oauthToken = "test_token_" + strconv.Itoa(testCount)
+			dummyTokenMan.token = testMock.oauthToken
+		})
+
+		AfterEach(func() {
+			<-testSnapMan.close()
+		})
+
+		It("downloadBootSnapshot happy path", func() {
+			testMock.normalAuthCheck()
+			testSnapMan.downloadBootSnapshot()
+			Expect(dummyDbMan.isDataSnapshot).Should(BeFalse())
+			Expect(dummyDbMan.snapshot.SnapshotInfo).Should(Equal(bootstrapSnapshotName))
+		})
+
+		It("downloadBootSnapshot should retry for auth failure", func() {
+			testMock.forceAuthFailOnce()
+			testSnapMan.downloadBootSnapshot()
+			Expect(dummyDbMan.isDataSnapshot).Should(BeFalse())
+			Expect(dummyDbMan.snapshot.SnapshotInfo).Should(Equal(bootstrapSnapshotName))
+			Expect(<-dummyTokenMan.invalidateChan).Should(BeTrue())
+		})
+
+		It("downloadDataSnapshot happy path", func() {
+			testMock.params.Scope = "test_scope_" + strconv.Itoa(testCount)
+			dummyDbMan.scopes = []string{testMock.params.Scope}
+			testMock.normalAuthCheck()
+			testSnapMan.downloadDataSnapshot()
+			Expect(dummyDbMan.isDataSnapshot).Should(BeTrue())
+			Expect(dummyDbMan.snapshot.SnapshotInfo).Should(Equal(testMock.snapshotID))
+		})
+
+		It("downloadDataSnapshot should retry for auth failure", func() {
+			testMock.params.Scope = "test_scope_" + strconv.Itoa(testCount)
+			dummyDbMan.scopes = []string{testMock.params.Scope}
+			testMock.forceAuthFailOnce()
+			testSnapMan.downloadDataSnapshot()
+			Expect(dummyDbMan.isDataSnapshot).Should(BeTrue())
+			Expect(dummyDbMan.snapshot.SnapshotInfo).Should(Equal(testMock.snapshotID))
+			Expect(<-dummyTokenMan.invalidateChan).Should(BeTrue())
+		})
+
+	})
+
+})
diff --git a/test_mock_test.go b/test_mock_test.go
index de2f673..9ba4e19 100644
--- a/test_mock_test.go
+++ b/test_mock_test.go
@@ -14,10 +14,115 @@
 package apidApigeeSync
 
 import (
+	"github.com/apid/apid-core"
 	"github.com/apigee-labs/transicator/common"
-	"net/url"
+	"math/rand"
+	"net/http"
+	"strconv"
 )
 
+type mockService struct {
+	config apid.ConfigService
+	log    apid.LogService
+	api    apid.APIService
+	data   apid.DataService
+	events apid.EventsService
+}
+
+func (s *mockService) API() apid.APIService {
+	return s.api
+}
+
+func (s *mockService) Config() apid.ConfigService {
+	return s.config
+}
+
+func (s *mockService) Data() apid.DataService {
+	return s.data
+}
+
+func (s *mockService) Events() apid.EventsService {
+	return s.events
+}
+
+func (s *mockService) Log() apid.LogService {
+	return s.log
+}
+
+type mockApi struct {
+	handleMap map[string]http.HandlerFunc
+}
+
+func (m *mockApi) Listen() error {
+	return nil
+}
+func (m *mockApi) Handle(path string, handler http.Handler) apid.Route {
+	return nil
+}
+func (m *mockApi) HandleFunc(path string, handlerFunc http.HandlerFunc) apid.Route {
+	m.handleMap[path] = handlerFunc
+	return apid.API().HandleFunc(path+strconv.Itoa(rand.Int()), handlerFunc)
+}
+func (m *mockApi) Vars(r *http.Request) map[string]string {
+	return nil
+}
+
+func (m *mockApi) Router() apid.Router {
+	return nil
+}
+
+type mockData struct {
+}
+
+func (m *mockData) DB() (apid.DB, error) {
+	return nil, nil
+}
+
+func (m *mockData) DBForID(id string) (apid.DB, error) {
+	return nil, nil
+}
+
+func (m *mockData) DBVersion(version string) (apid.DB, error) {
+	return nil, nil
+}
+func (m *mockData) DBVersionForID(id, version string) (apid.DB, error) {
+	return nil, nil
+}
+
+func (m *mockData) ReleaseDB(version string)          {}
+func (m *mockData) ReleaseCommonDB()                  {}
+func (m *mockData) ReleaseDBForID(id, version string) {}
+
+type mockEvent struct {
+	listenerMap map[apid.EventSelector]apid.EventHandlerFunc
+}
+
+func (e *mockEvent) Emit(selector apid.EventSelector, event apid.Event) chan apid.Event {
+	return nil
+}
+
+func (e *mockEvent) EmitWithCallback(selector apid.EventSelector, event apid.Event, handler apid.EventHandlerFunc) {
+
+}
+
+func (e *mockEvent) Listen(selector apid.EventSelector, handler apid.EventHandler) {
+
+}
+
+func (e *mockEvent) ListenFunc(selector apid.EventSelector, handler apid.EventHandlerFunc) {
+
+}
+
+func (e *mockEvent) ListenOnceFunc(selector apid.EventSelector, handler apid.EventHandlerFunc) {
+	e.listenerMap[selector] = handler
+}
+
+func (e *mockEvent) StopListening(selector apid.EventSelector, handler apid.EventHandler) {
+
+}
+
+func (e *mockEvent) Close() {}
+
 type dummyChangeManager struct {
 	pollChangeWithBackoffChan chan bool
 }
@@ -34,43 +139,34 @@
 
 type dummyTokenManager struct {
 	invalidateChan chan bool
+	token          string
+	tokenReadyChan chan bool
 }
 
 func (t *dummyTokenManager) getTokenReadyChannel() <-chan bool {
-	return nil
+	return t.tokenReadyChan
 }
 
 func (t *dummyTokenManager) getBearerToken() string {
-	return ""
+	return t.token
 }
 
-func (t *dummyTokenManager) invalidateToken() error {
+func (t *dummyTokenManager) invalidateToken() {
 	log.Debug("invalidateToken called")
-	testMock.passAuthCheck()
 	t.invalidateChan <- true
-	return nil
-}
-
-func (t *dummyTokenManager) getToken() *OauthToken {
-	return nil
 }
 
 func (t *dummyTokenManager) close() {
 	return
 }
 
-func (t *dummyTokenManager) getRetrieveNewTokenClosure(*url.URL) func(chan bool) error {
-	return func(chan bool) error {
-		return nil
-	}
-}
-
 func (t *dummyTokenManager) start() {
 
 }
 
 type dummySnapshotManager struct {
 	downloadCalledChan chan bool
+	startCalledChan    chan bool
 }
 
 func (s *dummySnapshotManager) close() <-chan bool {
@@ -80,28 +176,60 @@
 }
 
 func (s *dummySnapshotManager) downloadBootSnapshot() {
-
+	s.downloadCalledChan <- false
 }
 
-func (s *dummySnapshotManager) storeBootSnapshot(snapshot *common.Snapshot) {
-
-}
-
-func (s *dummySnapshotManager) downloadDataSnapshot() {
-	log.Debug("dummySnapshotManager.downloadDataSnapshot() called")
+func (s *dummySnapshotManager) downloadDataSnapshot() error {
 	s.downloadCalledChan <- true
-}
-
-func (s *dummySnapshotManager) storeDataSnapshot(snapshot *common.Snapshot) {
-
-}
-
-func (s *dummySnapshotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error {
 	return nil
 }
 
-func (s *dummySnapshotManager) startOnLocalSnapshot(snapshot string) *common.Snapshot {
-	return &common.Snapshot{
-		SnapshotInfo: snapshot,
-	}
+func (s *dummySnapshotManager) startOnDataSnapshot(snapshot string) error {
+	s.startCalledChan <- true
+	return nil
+}
+
+type dummyDbManager struct {
+	lastSequence   string
+	knownTables    map[string]bool
+	scopes         []string
+	snapshot       *common.Snapshot
+	isDataSnapshot bool
+	lastSeqUpdated chan string
+}
+
+func (d *dummyDbManager) initDB() error {
+	return nil
+}
+func (d *dummyDbManager) setDB(db apid.DB) {
+
+}
+func (d *dummyDbManager) getLastSequence() (lastSequence string) {
+	return d.lastSequence
+}
+func (d *dummyDbManager) findScopesForId(configId string) (scopes []string, err error) {
+	return d.scopes, nil
+}
+func (d *dummyDbManager) updateLastSequence(lastSequence string) error {
+	d.lastSeqUpdated <- lastSequence
+	return nil
+}
+func (d *dummyDbManager) getApidInstanceInfo() (info apidInstanceInfo, err error) {
+	return apidInstanceInfo{
+		InstanceID:   "",
+		InstanceName: "",
+		ClusterID:    "",
+		LastSnapshot: "",
+	}, nil
+}
+func (d *dummyDbManager) processChangeList(changes *common.ChangeList) error {
+	return nil
+}
+func (d *dummyDbManager) processSnapshot(snapshot *common.Snapshot, isDataSnapshot bool) error {
+	d.snapshot = snapshot
+	d.isDataSnapshot = isDataSnapshot
+	return nil
+}
+func (d *dummyDbManager) getKnowTables() map[string]bool {
+	return d.knownTables
 }
diff --git a/token.go b/token.go
index 95d4b06..c355a61 100644
--- a/token.go
+++ b/token.go
@@ -17,7 +17,6 @@
 import (
 	"bytes"
 	"encoding/json"
-	"errors"
 	"github.com/apid/apid-core/util"
 	"io/ioutil"
 	"net/http"
@@ -35,15 +34,13 @@
 Usage:
    man := createTokenManager()
    bearer := man.getBearerToken()
-   // will automatically update config(configBearerToken) for other modules
-   // optionally, when done...
-   man.close()
+   will automatically update config(configBearerToken) for other modules
 */
 
-func createSimpleTokenManager() *simpleTokenManager {
+func createApidTokenManager(isNewInstance bool) *apidTokenManager {
 	isClosedInt := int32(0)
 
-	t := &simpleTokenManager{
+	t := &apidTokenManager{
 		quitPollingForToken: make(chan bool, 1),
 		closed:              make(chan bool),
 		getTokenChan:        make(chan bool),
@@ -52,11 +49,12 @@
 		invalidateDone:      make(chan bool),
 		tokenUpdatedChan:    make(chan bool, 1),
 		isClosed:            &isClosedInt,
+		isNewInstance:       isNewInstance,
 	}
 	return t
 }
 
-type simpleTokenManager struct {
+type apidTokenManager struct {
 	token               *OauthToken
 	isClosed            *int32
 	quitPollingForToken chan bool
@@ -67,19 +65,20 @@
 	returnTokenChan     chan *OauthToken
 	invalidateDone      chan bool
 	tokenUpdatedChan    chan bool
+	isNewInstance       bool
 }
 
-func (t *simpleTokenManager) start() {
+func (t *apidTokenManager) start() {
 	t.retrieveNewToken()
 	t.refreshTimer = time.After(t.token.refreshIn())
 	go t.maintainToken()
 }
 
-func (t *simpleTokenManager) getBearerToken() string {
+func (t *apidTokenManager) getBearerToken() string {
 	return t.getToken().AccessToken
 }
 
-func (t *simpleTokenManager) maintainToken() {
+func (t *apidTokenManager) maintainToken() {
 	for {
 		select {
 		case <-t.closed:
@@ -100,19 +99,13 @@
 }
 
 // will block until valid
-func (t *simpleTokenManager) invalidateToken() error {
-	//has been closed
-	if atomic.LoadInt32(t.isClosed) == int32(1) {
-		log.Debug("TokenManager: invalidateToken() called on closed tokenManager")
-		return errors.New("invalidateToken() called on closed tokenManager")
-	}
+func (t *apidTokenManager) invalidateToken() {
 	log.Debug("invalidating token")
 	t.invalidateTokenChan <- true
 	<-t.invalidateDone
-	return nil
 }
 
-func (t *simpleTokenManager) getToken() *OauthToken {
+func (t *apidTokenManager) getToken() *OauthToken {
 	//has been closed
 	if atomic.LoadInt32(t.isClosed) == int32(1) {
 		log.Debug("TokenManager: getToken() called on closed tokenManager")
@@ -126,7 +119,7 @@
  * blocking close() of tokenMan
  */
 
-func (t *simpleTokenManager) close() {
+func (t *apidTokenManager) close() {
 	//has been closed
 	if atomic.SwapInt32(t.isClosed, 1) == int32(1) {
 		log.Panic("TokenManager: close() has been called before!")
@@ -141,7 +134,7 @@
 }
 
 // don't call externally. will block until success.
-func (t *simpleTokenManager) retrieveNewToken() {
+func (t *apidTokenManager) retrieveNewToken() {
 
 	log.Debug("Getting OAuth token...")
 	uriString := config.GetString(configProxyServerBaseURI)
@@ -151,10 +144,10 @@
 	}
 	uri.Path = path.Join(uri.Path, "/accesstoken")
 
-	pollWithBackoff(t.quitPollingForToken, t.getRetrieveNewTokenClosure(uri), func(err error) { log.Errorf("Error getting new token : ", err) })
+	pollWithBackoff(t.quitPollingForToken, t.getRetrieveNewTokenClosure(uri), func(err error) { log.Errorf("Error getting new token : %v", err) })
 }
 
-func (t *simpleTokenManager) getRetrieveNewTokenClosure(uri *url.URL) func(chan bool) error {
+func (t *apidTokenManager) getRetrieveNewTokenClosure(uri *url.URL) func(chan bool) error {
 	return func(_ chan bool) error {
 		form := url.Values{}
 		form.Set("grant_type", "client_credentials")
@@ -168,9 +161,9 @@
 		req.Header.Set("status", "ONLINE")
 		req.Header.Set("plugin_details", apidPluginDetails)
 
-		if newInstanceID {
+		if t.isNewInstance {
 			req.Header.Set("created_at_apid", time.Now().Format(time.RFC3339))
-			newInstanceID = false
+			t.isNewInstance = false
 		} else {
 			req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
 		}
@@ -194,7 +187,7 @@
 
 		if resp.StatusCode != 200 {
 			log.Errorf("Oauth Request Failed with Resp Code: %d. Body: %s", resp.StatusCode, string(body))
-			return expected200Error{}
+			return expected200Error
 		}
 
 		var token OauthToken
@@ -212,22 +205,11 @@
 		}
 
 		log.Debugf("Got new token: %#v", token)
-
-		/*
-			if newInstanceID {
-				newInstanceID = false
-				err = updateApidInstanceInfo()
-				if err != nil {
-					log.Errorf("unable to unmarshal update apid instance info : %v", string(body), err)
-					return err
-
-				}
-			}
-		*/
 		t.token = &token
 		config.Set(configBearerToken, token.AccessToken)
 
 		//don't block on the buffered channel.  that means there is already a signal to serve new token
+		//TODO: This assumes apid-gateway is 1-1 mapping. Make use of generic long-polling provided by apid-core
 		select {
 		case t.tokenUpdatedChan <- true:
 		default:
@@ -238,7 +220,7 @@
 	}
 }
 
-func (t *simpleTokenManager) getTokenReadyChannel() <-chan bool {
+func (t *apidTokenManager) getTokenReadyChannel() <-chan bool {
 	return t.tokenUpdatedChan
 }
 
diff --git a/token_test.go b/token_test.go
index 1dfaabb..fa9379c 100644
--- a/token_test.go
+++ b/token_test.go
@@ -94,7 +94,7 @@
 				w.Write(body)
 			}))
 			config.Set(configProxyServerBaseURI, ts.URL)
-			testedTokenManager := createSimpleTokenManager()
+			testedTokenManager := createApidTokenManager(false)
 			testedTokenManager.start()
 			token := testedTokenManager.getToken()
 
@@ -123,7 +123,7 @@
 			}))
 			config.Set(configProxyServerBaseURI, ts.URL)
 
-			testedTokenManager := createSimpleTokenManager()
+			testedTokenManager := createApidTokenManager(false)
 			testedTokenManager.start()
 			token := testedTokenManager.getToken()
 			Expect(token.AccessToken).ToNot(BeEmpty())
@@ -163,7 +163,7 @@
 			}))
 
 			config.Set(configProxyServerBaseURI, ts.URL)
-			testedTokenManager := createSimpleTokenManager()
+			testedTokenManager := createApidTokenManager(false)
 			testedTokenManager.start()
 			testedTokenManager.getToken()
 
@@ -179,8 +179,6 @@
 			finished := make(chan bool, 1)
 			count := 0
 
-			newInstanceID = true
-
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 
 				count++
@@ -202,7 +200,7 @@
 			}))
 
 			config.Set(configProxyServerBaseURI, ts.URL)
-			testedTokenManager := createSimpleTokenManager()
+			testedTokenManager := createApidTokenManager(true)
 			testedTokenManager.start()
 			testedTokenManager.getToken()
 			testedTokenManager.invalidateToken()
diff --git a/util.go b/util.go
new file mode 100644
index 0000000..7d59214
--- /dev/null
+++ b/util.go
@@ -0,0 +1,172 @@
+// Copyright 2017 Google Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apidApigeeSync
+
+import (
+	"fmt"
+	"math"
+	"math/rand"
+	"net/http"
+	"time"
+)
+
+const (
+	httpTimeout                       = time.Minute
+	pluginTimeout                     = time.Minute
+	maxIdleConnsPerHost               = 10
+	defaultInitial      time.Duration = 200 * time.Millisecond
+	defaultMax          time.Duration = 10 * time.Second
+	defaultFactor       float64       = 2
+)
+
+var (
+	initialBackoffInterval = defaultInitial
+)
+
+var (
+	expected200Error = fmt.Errorf("did not recieve OK response")
+	quitSignalError  = fmt.Errorf("signal to quit encountered")
+	authFailError    = fmt.Errorf("authorization failed")
+)
+
+type Backoff struct {
+	attempt         int
+	initial, max    time.Duration
+	jitter          bool
+	backoffStrategy func() time.Duration
+}
+
+type ExponentialBackoff struct {
+	Backoff
+	factor float64
+}
+
+func NewExponentialBackoff(initial, max time.Duration, factor float64, jitter bool) *ExponentialBackoff {
+	backoff := &ExponentialBackoff{}
+
+	if initial <= 0 {
+		initial = defaultInitial
+	}
+	if max <= 0 {
+		max = defaultMax
+	}
+
+	if factor <= 0 {
+		factor = defaultFactor
+	}
+
+	backoff.initial = initial
+	backoff.max = max
+	backoff.attempt = 0
+	backoff.factor = factor
+	backoff.jitter = jitter
+	backoff.backoffStrategy = backoff.exponentialBackoffStrategy
+
+	return backoff
+}
+
+func (b *Backoff) Duration() time.Duration {
+	d := b.backoffStrategy()
+	b.attempt++
+	return d
+}
+
+func (b *ExponentialBackoff) exponentialBackoffStrategy() time.Duration {
+
+	initial := float64(b.Backoff.initial)
+	attempt := float64(b.Backoff.attempt)
+	duration := initial * math.Pow(b.factor, attempt)
+
+	if duration > math.MaxInt64 {
+		return b.max
+	}
+	dur := time.Duration(duration)
+
+	if b.jitter {
+		duration = rand.Float64()*(duration-initial) + initial
+	}
+
+	if dur > b.max {
+		return b.max
+	}
+
+	log.Debugf("Backing off for %d ms", int64(dur/time.Millisecond))
+	return dur
+}
+
+func (b *Backoff) Reset() {
+	b.attempt = 0
+}
+
+func (b *Backoff) Attempt() int {
+	return b.attempt
+}
+
+/*
+ * Call toExecute repeatedly until it does not return an error, with an exponential backoff policy
+ * for retrying on errors
+ */
+func pollWithBackoff(quit chan bool, toExecute func(chan bool) error, handleError func(error)) {
+
+	backoff := NewExponentialBackoff(initialBackoffInterval, config.GetDuration(configPollInterval), 2, true)
+
+	//initialize the retry channel to start first attempt immediately
+	retry := time.After(0 * time.Millisecond)
+
+	for {
+		select {
+		case <-quit:
+			log.Info("Quit signal recieved.  Returning")
+			return
+		case <-retry:
+			start := time.Now()
+
+			err := toExecute(quit)
+			if err == nil || err == quitSignalError {
+				return
+			}
+
+			end := time.Now()
+			//error encountered, since we would have returned above otherwise
+			handleError(err)
+
+			/* TODO keep this around? Imagine an immediately erroring service,
+			 *  causing many sequential requests which could pollute logs
+			 */
+			//only backoff if the request took less than one second
+			if end.After(start.Add(time.Second)) {
+				backoff.Reset()
+				retry = time.After(0 * time.Millisecond)
+			} else {
+				retry = time.After(backoff.Duration())
+			}
+		}
+	}
+}
+
+func addHeaders(req *http.Request, token string) {
+	req.Header.Set("Authorization", "Bearer "+token)
+	req.Header.Set("apid_instance_id", apidInfo.InstanceID)
+	req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
+	req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
+}
+
+type changeServerError struct {
+	Code string `json:"code"`
+}
+
+func (a changeServerError) Error() string {
+	return a.Code
+}
diff --git a/backoff_test.go b/util_test.go
similarity index 100%
rename from backoff_test.go
rename to util_test.go