killbill-uncached

diff --git a/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java b/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
index caefaef..eb41bce 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/AnalyticsListener.java
@@ -21,20 +21,37 @@ import com.google.inject.Inject;
 import com.ning.billing.account.api.AccountApiException;
 import com.ning.billing.account.api.AccountChangeEvent;
 import com.ning.billing.account.api.AccountCreationEvent;
+import com.ning.billing.entitlement.api.timeline.RepairEntitlementEvent;
 import com.ning.billing.entitlement.api.user.EntitlementUserApiException;
 import com.ning.billing.entitlement.api.user.SubscriptionEvent;
+import com.ning.billing.invoice.api.EmptyInvoiceEvent;
 import com.ning.billing.invoice.api.InvoiceCreationEvent;
 import com.ning.billing.payment.api.PaymentErrorEvent;
 import com.ning.billing.payment.api.PaymentInfoEvent;
+import com.ning.billing.util.tag.api.ControlTagCreationEvent;
+import com.ning.billing.util.tag.api.ControlTagDefinitionCreationEvent;
+import com.ning.billing.util.tag.api.ControlTagDefinitionDeletionEvent;
+import com.ning.billing.util.tag.api.ControlTagDeletionEvent;
+import com.ning.billing.util.tag.api.UserTagCreationEvent;
+import com.ning.billing.util.tag.api.UserTagDefinitionCreationEvent;
+import com.ning.billing.util.tag.api.UserTagDefinitionDeletionEvent;
+import com.ning.billing.util.tag.api.UserTagDeletionEvent;
 
 public class AnalyticsListener {
     private final BusinessSubscriptionTransitionRecorder bstRecorder;
     private final BusinessAccountRecorder bacRecorder;
+    private final BusinessInvoiceRecorder invoiceRecorder;
+    private final BusinessTagRecorder tagRecorder;
 
     @Inject
-    public AnalyticsListener(final BusinessSubscriptionTransitionRecorder bstRecorder, final BusinessAccountRecorder bacRecorder) {
+    public AnalyticsListener(final BusinessSubscriptionTransitionRecorder bstRecorder,
+                             final BusinessAccountRecorder bacRecorder,
+                             final BusinessInvoiceRecorder invoiceRecorder,
+                             final BusinessTagRecorder tagRecorder) {
         this.bstRecorder = bstRecorder;
         this.bacRecorder = bacRecorder;
+        this.invoiceRecorder = invoiceRecorder;
+        this.tagRecorder = tagRecorder;
     }
 
     @Subscribe
@@ -77,12 +94,17 @@ public class AnalyticsListener {
             return;
         }
 
-        bacRecorder.accountUpdated(event.getAccountId(), event.getChangedFields());
+        bacRecorder.accountUpdated(event.getAccountId());
     }
 
     @Subscribe
-    public void handleInvoice(final InvoiceCreationEvent event) {
-        bacRecorder.accountUpdated(event.getAccountId());
+    public void handleInvoiceCreation(final InvoiceCreationEvent event) {
+        invoiceRecorder.invoiceCreated(event.getInvoiceId());
+    }
+
+    @Subscribe
+    public void handleNullInvoice(final EmptyInvoiceEvent event) {
+        // Ignored for now
     }
 
     @Subscribe
@@ -94,4 +116,49 @@ public class AnalyticsListener {
     public void handlePaymentError(final PaymentErrorEvent paymentError) {
         // TODO - we can't tie the error back to an account yet
     }
+
+    @Subscribe
+    public void handleControlTagCreation(final ControlTagCreationEvent event) {
+        tagRecorder.tagAdded(event.getObjectType(), event.getObjectId(), event.getTagDefinition().getName());
+    }
+
+    @Subscribe
+    public void handleControlTagDeletion(final ControlTagDeletionEvent event) {
+        tagRecorder.tagRemoved(event.getObjectType(), event.getObjectId(), event.getTagDefinition().getName());
+    }
+
+    @Subscribe
+    public void handleUserTagCreation(final UserTagCreationEvent event) {
+        tagRecorder.tagAdded(event.getObjectType(), event.getObjectId(), event.getTagDefinition().getName());
+    }
+
+    @Subscribe
+    public void handleUserTagDeletion(final UserTagDeletionEvent event) {
+        tagRecorder.tagRemoved(event.getObjectType(), event.getObjectId(), event.getTagDefinition().getName());
+    }
+
+    @Subscribe
+    public void handleControlTagDefinitionCreation(final ControlTagDefinitionCreationEvent event) {
+        // Ignored for now
+    }
+
+    @Subscribe
+    public void handleControlTagDefinitionDeletion(final ControlTagDefinitionDeletionEvent event) {
+        // Ignored for now
+    }
+
+    @Subscribe
+    public void handleUserTagDefinitionCreation(final UserTagDefinitionCreationEvent event) {
+        // Ignored for now
+    }
+
+    @Subscribe
+    public void handleUserTagDefinitionDeletion(final UserTagDefinitionDeletionEvent event) {
+        // Ignored for now
+    }
+
+    @Subscribe
+    public void handleRepairEntitlement(final RepairEntitlementEvent event) {
+        // Ignored for now
+    }
 }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/api/user/DefaultAnalyticsUserApi.java b/analytics/src/main/java/com/ning/billing/analytics/api/user/DefaultAnalyticsUserApi.java
new file mode 100644
index 0000000..424e1c2
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/api/user/DefaultAnalyticsUserApi.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.api.user;
+
+import javax.inject.Inject;
+import java.util.List;
+import java.util.UUID;
+
+import com.ning.billing.analytics.dao.AnalyticsDao;
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessAccountTag;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
+
+// Note: not exposed in api yet
+public class DefaultAnalyticsUserApi {
+    private final AnalyticsDao analyticsDao;
+
+    @Inject
+    public DefaultAnalyticsUserApi(final AnalyticsDao analyticsDao) {
+        this.analyticsDao = analyticsDao;
+    }
+
+    public BusinessAccount getAccountByKey(final String accountKey) {
+        return analyticsDao.getAccountByKey(accountKey);
+    }
+
+    public List<BusinessSubscriptionTransition> getTransitionsForBundle(final String externalKey) {
+        return analyticsDao.getTransitionsByKey(externalKey);
+    }
+
+    public List<BusinessInvoice> getInvoicesForAccount(final String accountKey) {
+        return analyticsDao.getInvoicesByKey(accountKey);
+    }
+
+    public List<BusinessAccountTag> getTagsForAccount(final String accountKey) {
+        return analyticsDao.getTagsForAccount(accountKey);
+    }
+
+    public List<BusinessInvoiceItem> getInvoiceItemsForInvoice(final UUID invoiceId) {
+        return analyticsDao.getInvoiceItemsForInvoice(invoiceId.toString());
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java b/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java
index 7f69fef..998fd7e 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java
@@ -17,9 +17,7 @@
 package com.ning.billing.analytics;
 
 import java.math.BigDecimal;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
@@ -31,7 +29,6 @@ import com.ning.billing.account.api.Account;
 import com.ning.billing.account.api.AccountApiException;
 import com.ning.billing.account.api.AccountData;
 import com.ning.billing.account.api.AccountUserApi;
-import com.ning.billing.account.api.ChangedField;
 import com.ning.billing.analytics.dao.BusinessAccountSqlDao;
 import com.ning.billing.analytics.model.BusinessAccount;
 import com.ning.billing.invoice.api.Invoice;
@@ -40,9 +37,6 @@ import com.ning.billing.payment.api.Payment;
 import com.ning.billing.payment.api.PaymentApi;
 import com.ning.billing.payment.api.PaymentApiException;
 import com.ning.billing.payment.api.PaymentInfoEvent;
-import com.ning.billing.util.api.TagUserApi;
-import com.ning.billing.util.dao.ObjectType;
-import com.ning.billing.util.tag.Tag;
 
 public class BusinessAccountRecorder {
     private static final Logger log = LoggerFactory.getLogger(BusinessAccountRecorder.class);
@@ -51,25 +45,22 @@ public class BusinessAccountRecorder {
     private final AccountUserApi accountApi;
     private final InvoiceUserApi invoiceUserApi;
     private final PaymentApi paymentApi;
-    private final TagUserApi tagUserApi;
 
     @Inject
     public BusinessAccountRecorder(final BusinessAccountSqlDao sqlDao, final AccountUserApi accountApi,
-                                   final InvoiceUserApi invoiceUserApi, final PaymentApi paymentApi,
-                                   final TagUserApi tagUserApi) {
+                                   final InvoiceUserApi invoiceUserApi, final PaymentApi paymentApi) {
         this.sqlDao = sqlDao;
         this.accountApi = accountApi;
         this.invoiceUserApi = invoiceUserApi;
         this.paymentApi = paymentApi;
-        this.tagUserApi = tagUserApi;
     }
 
     public void accountCreated(final AccountData data) {
         final Account account;
         try {
             account = accountApi.getAccountByKey(data.getExternalKey());
-            final Map<String, Tag> tags = tagUserApi.getTags(account.getId(), ObjectType.ACCOUNT);
-            final BusinessAccount bac = createBusinessAccountFromAccount(account, new ArrayList<Tag>(tags.values()));
+            final BusinessAccount bac = new BusinessAccount();
+            updateBusinessAccountFromAccount(account, bac);
 
             log.info("ACCOUNT CREATION " + bac);
             sqlDao.createAccount(bac);
@@ -79,17 +70,6 @@ public class BusinessAccountRecorder {
     }
 
     /**
-     * Notification handler for Account changes
-     *
-     * @param accountId     account id changed
-     * @param changedFields list of changed fields
-     */
-    public void accountUpdated(final UUID accountId, final List<ChangedField> changedFields) {
-        // None of the fields updated interest us so far - see DefaultAccountChangeNotification
-        // TODO We'll need notifications for tags changes eventually
-    }
-
-    /**
      * Notification handler for Payment creations
      *
      * @param paymentInfo payment object (from the payment plugin)
@@ -111,16 +91,11 @@ public class BusinessAccountRecorder {
     public void accountUpdated(final UUID accountId) {
         try {
             final Account account = accountApi.getAccountById(accountId);
-            final Map<String, Tag> tags = tagUserApi.getTags(accountId, ObjectType.ACCOUNT);
-
-            if (account == null) {
-                log.warn("Couldn't find account {}", accountId);
-                return;
-            }
 
             BusinessAccount bac = sqlDao.getAccount(account.getExternalKey());
             if (bac == null) {
-                bac = createBusinessAccountFromAccount(account, new ArrayList<Tag>(tags.values()));
+                bac = new BusinessAccount();
+                updateBusinessAccountFromAccount(account, bac);
                 log.info("ACCOUNT CREATION " + bac);
                 sqlDao.createAccount(bac);
             } else {
@@ -131,44 +106,25 @@ public class BusinessAccountRecorder {
         } catch (AccountApiException e) {
             log.warn("Error encountered creating BusinessAccount", e);
         }
-
-    }
-
-    private BusinessAccount createBusinessAccountFromAccount(final Account account, final List<Tag> tags) {
-        final BusinessAccount bac = new BusinessAccount(
-                account.getExternalKey(),
-                account.getName(),
-                invoiceUserApi.getAccountBalance(account.getId()),
-                // These fields will be updated below
-                null,
-                null,
-                null,
-                null,
-                null,
-                null
-        );
-        updateBusinessAccountFromAccount(account, bac);
-
-        return bac;
     }
 
     private void updateBusinessAccountFromAccount(final Account account, final BusinessAccount bac) {
+        bac.setName(account.getName());
+        bac.setKey(account.getExternalKey());
 
-        final List<UUID> invoiceIds = new ArrayList<UUID>();
         try {
-            DateTime lastInvoiceDate = null;
-            BigDecimal totalInvoiceBalance = BigDecimal.ZERO;
-            String lastPaymentStatus = null;
-            String paymentMethod = null;
-            String creditCardType = null;
-            String billingAddressCountry = null;
+            DateTime lastInvoiceDate = bac.getLastInvoiceDate();
+            BigDecimal totalInvoiceBalance = bac.getTotalInvoiceBalance();
+            String lastPaymentStatus = bac.getLastPaymentStatus();
+            String paymentMethod = bac.getPaymentMethod();
+            String creditCardType = bac.getCreditCardType();
+            String billingAddressCountry = bac.getBillingAddressCountry();
 
             // Retrieve invoices information
             final List<Invoice> invoices = invoiceUserApi.getInvoicesByAccount(account.getId());
             if (invoices != null && invoices.size() > 0) {
 
                 for (final Invoice invoice : invoices) {
-                    invoiceIds.add(invoice.getId());
                     totalInvoiceBalance = totalInvoiceBalance.add(invoice.getBalance());
 
                     if (lastInvoiceDate == null || invoice.getInvoiceDate().isAfter(lastInvoiceDate)) {

diff --git a/analytics/src/main/java/com/ning/billing/analytics/BusinessInvoiceRecorder.java b/analytics/src/main/java/com/ning/billing/analytics/BusinessInvoiceRecorder.java
new file mode 100644
index 0000000..21693f3
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/BusinessInvoiceRecorder.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics;
+
+import javax.inject.Inject;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.ning.billing.account.api.Account;
+import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.account.api.AccountUserApi;
+import com.ning.billing.analytics.dao.AnalyticsDao;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.catalog.api.Plan;
+import com.ning.billing.catalog.api.PlanPhase;
+import com.ning.billing.entitlement.api.user.EntitlementUserApi;
+import com.ning.billing.entitlement.api.user.EntitlementUserApiException;
+import com.ning.billing.entitlement.api.user.Subscription;
+import com.ning.billing.entitlement.api.user.SubscriptionBundle;
+import com.ning.billing.invoice.api.Invoice;
+import com.ning.billing.invoice.api.InvoiceItem;
+import com.ning.billing.invoice.api.InvoiceUserApi;
+
+public class BusinessInvoiceRecorder {
+    private static final Logger log = LoggerFactory.getLogger(BusinessInvoiceRecorder.class);
+
+    private final AnalyticsDao analyticsDao;
+    private final AccountUserApi accountApi;
+    private final EntitlementUserApi entitlementApi;
+    private final InvoiceUserApi invoiceApi;
+
+    @Inject
+    public BusinessInvoiceRecorder(final AnalyticsDao analyticsDao,
+                                   final AccountUserApi accountApi,
+                                   final EntitlementUserApi entitlementApi,
+                                   final InvoiceUserApi invoiceApi) {
+        this.analyticsDao = analyticsDao;
+        this.accountApi = accountApi;
+        this.entitlementApi = entitlementApi;
+        this.invoiceApi = invoiceApi;
+    }
+
+    public void invoiceCreated(final UUID invoiceId) {
+        // Lookup the invoice object
+        final Invoice invoice = invoiceApi.getInvoice(invoiceId);
+        if (invoice == null) {
+            log.warn("Ignoring invoice creation for invoice id {} (invoice does not exist)", invoiceId.toString());
+            return;
+        }
+
+        // Lookup the associated account
+        final String accountKey;
+        try {
+            final Account account = accountApi.getAccountById(invoice.getAccountId());
+            accountKey = account.getExternalKey();
+        } catch (AccountApiException e) {
+            log.warn("Ignoring invoice creation for invoice id {} and account id {} (account does not exist)",
+                     invoice.getId().toString(),
+                     invoice.getAccountId().toString());
+            return;
+        }
+
+        // Create the invoice
+        final BusinessInvoice businessInvoice = new BusinessInvoice(accountKey, invoice);
+
+        // Create associated invoice items
+        final List<BusinessInvoiceItem> businessInvoiceItems = new ArrayList<BusinessInvoiceItem>();
+        for (final InvoiceItem invoiceItem : invoice.getInvoiceItems()) {
+            final BusinessInvoiceItem businessInvoiceItem = createBusinessInvoiceItem(invoiceItem);
+            if (businessInvoiceItem != null) {
+                businessInvoiceItems.add(businessInvoiceItem);
+            }
+        }
+
+        // Update the Analytics tables
+        analyticsDao.createInvoice(accountKey, businessInvoice, businessInvoiceItems);
+    }
+
+    private BusinessInvoiceItem createBusinessInvoiceItem(final InvoiceItem invoiceItem) {
+        final String externalKey;
+        try {
+            final SubscriptionBundle bundle = entitlementApi.getBundleFromId(invoiceItem.getBundleId());
+            externalKey = bundle.getKey();
+        } catch (EntitlementUserApiException e) {
+            log.warn("Ignoring invoice item {} for bundle {} (bundle does not exist)",
+                     invoiceItem.getId().toString(),
+                     invoiceItem.getBundleId().toString());
+            return null;
+        }
+
+        final Subscription subscription;
+        try {
+            subscription = entitlementApi.getSubscriptionFromId(invoiceItem.getSubscriptionId());
+        } catch (EntitlementUserApiException e) {
+            log.warn("Ignoring invoice item {} for subscription {} (subscription does not exist)",
+                     invoiceItem.getId().toString(),
+                     invoiceItem.getSubscriptionId().toString());
+            return null;
+        }
+
+        final Plan plan = subscription.getCurrentPlan();
+        if (plan == null) {
+            log.warn("Ignoring invoice item {} for subscription {} (null plan)",
+                     invoiceItem.getId().toString(),
+                     invoiceItem.getSubscriptionId().toString());
+            return null;
+        }
+
+        final PlanPhase planPhase = subscription.getCurrentPhase();
+        if (planPhase == null) {
+            log.warn("Ignoring invoice item {} for subscription {} (null phase)",
+                     invoiceItem.getId().toString(),
+                     invoiceItem.getSubscriptionId().toString());
+            return null;
+        }
+
+        return new BusinessInvoiceItem(externalKey, invoiceItem, plan, planPhase);
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/BusinessTagRecorder.java b/analytics/src/main/java/com/ning/billing/analytics/BusinessTagRecorder.java
new file mode 100644
index 0000000..8aa4343
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/BusinessTagRecorder.java
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics;
+
+import javax.inject.Inject;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.ning.billing.account.api.Account;
+import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.account.api.AccountUserApi;
+import com.ning.billing.analytics.dao.BusinessAccountTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoicePaymentTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionTagSqlDao;
+import com.ning.billing.entitlement.api.user.EntitlementUserApi;
+import com.ning.billing.entitlement.api.user.EntitlementUserApiException;
+import com.ning.billing.entitlement.api.user.SubscriptionBundle;
+import com.ning.billing.util.dao.ObjectType;
+
+public class BusinessTagRecorder {
+    private static final Logger log = LoggerFactory.getLogger(BusinessTagRecorder.class);
+
+    private final BusinessAccountTagSqlDao accountTagSqlDao;
+    private final BusinessInvoiceTagSqlDao invoiceTagSqlDao;
+    private final BusinessInvoicePaymentTagSqlDao invoicePaymentTagSqlDao;
+    private final BusinessSubscriptionTransitionTagSqlDao subscriptionTransitionTagSqlDao;
+    private final AccountUserApi accountApi;
+    private final EntitlementUserApi entitlementUserApi;
+
+    @Inject
+    public BusinessTagRecorder(final BusinessAccountTagSqlDao accountTagSqlDao,
+                               final BusinessInvoicePaymentTagSqlDao invoicePaymentTagSqlDao,
+                               final BusinessInvoiceTagSqlDao invoiceTagSqlDao,
+                               final BusinessSubscriptionTransitionTagSqlDao subscriptionTransitionTagSqlDao,
+                               final AccountUserApi accountApi,
+                               final EntitlementUserApi entitlementUserApi) {
+        this.accountTagSqlDao = accountTagSqlDao;
+        this.invoicePaymentTagSqlDao = invoicePaymentTagSqlDao;
+        this.invoiceTagSqlDao = invoiceTagSqlDao;
+        this.subscriptionTransitionTagSqlDao = subscriptionTransitionTagSqlDao;
+        this.accountApi = accountApi;
+        this.entitlementUserApi = entitlementUserApi;
+    }
+
+    public void tagAdded(final ObjectType objectType, final UUID objectId, final String name) {
+        if (objectType.equals(ObjectType.ACCOUNT)) {
+            tagAddedForAccount(objectId, name);
+        } else if (objectType.equals(ObjectType.BUNDLE)) {
+            tagAddedForBundle(objectId, name);
+        } else if (objectType.equals(ObjectType.INVOICE)) {
+            tagAddedForInvoice(objectId, name);
+        } else if (objectType.equals(ObjectType.PAYMENT)) {
+            tagAddedForPayment(objectId, name);
+        } else {
+            log.info("Ignoring tag addition of {} for object id {} (type {})", new Object[]{name, objectId.toString(), objectType.toString()});
+        }
+    }
+
+    public void tagRemoved(final ObjectType objectType, final UUID objectId, final String name) {
+        if (objectType.equals(ObjectType.ACCOUNT)) {
+            tagRemovedForAccount(objectId, name);
+        } else if (objectType.equals(ObjectType.BUNDLE)) {
+            tagRemovedForBundle(objectId, name);
+        } else if (objectType.equals(ObjectType.INVOICE)) {
+            tagRemovedForInvoice(objectId, name);
+        } else if (objectType.equals(ObjectType.PAYMENT)) {
+            tagRemovedForPayment(objectId, name);
+        } else {
+            log.info("Ignoring tag removal of {} for object id {} (type {})", new Object[]{name, objectId.toString(), objectType.toString()});
+        }
+    }
+
+    private void tagAddedForAccount(final UUID objectId, final String name) {
+        final Account account;
+        try {
+            account = accountApi.getAccountById(objectId);
+        } catch (AccountApiException e) {
+            log.warn("Ignoring tag addition of {} for account id {} (account does not exist)", name, objectId.toString());
+            return;
+        }
+
+        final String accountKey = account.getExternalKey();
+        accountTagSqlDao.addTag(accountKey, name);
+    }
+
+    private void tagRemovedForAccount(final UUID objectId, final String name) {
+        final Account account;
+        try {
+            account = accountApi.getAccountById(objectId);
+        } catch (AccountApiException e) {
+            log.warn("Ignoring tag removal of {} for account id {} (account does not exist)", name, objectId.toString());
+            return;
+        }
+
+        final String accountKey = account.getExternalKey();
+        accountTagSqlDao.removeTag(accountKey, name);
+    }
+
+    private void tagAddedForBundle(final UUID objectId, final String name) {
+        final SubscriptionBundle bundle;
+        try {
+            bundle = entitlementUserApi.getBundleFromId(objectId);
+        } catch (EntitlementUserApiException e) {
+            log.warn("Ignoring tag addition of {} for bundle id {} (bundle does not exist)", name, objectId.toString());
+            return;
+        }
+
+        /*
+         * Note: we store tags associated to bundles, not to subscriptions.
+         * Subscriptions are in the core of killbill and not exposed in Analytics to avoid a hard dependency
+         * (i.e. dashboards should not rely on killbill ids).
+         */
+        final String externalKey = bundle.getKey();
+        subscriptionTransitionTagSqlDao.addTag(externalKey, name);
+    }
+
+    private void tagRemovedForBundle(final UUID objectId, final String name) {
+        final SubscriptionBundle bundle;
+        try {
+            bundle = entitlementUserApi.getBundleFromId(objectId);
+        } catch (EntitlementUserApiException e) {
+            log.warn("Ignoring tag removal of {} for bundle id {} (bundle does not exist)", name, objectId.toString());
+            return;
+        }
+
+        final String externalKey = bundle.getKey();
+        subscriptionTransitionTagSqlDao.removeTag(externalKey, name);
+    }
+
+    private void tagAddedForInvoice(final UUID objectId, final String name) {
+        invoiceTagSqlDao.addTag(objectId.toString(), name);
+    }
+
+    private void tagRemovedForInvoice(final UUID objectId, final String name) {
+        invoiceTagSqlDao.removeTag(objectId.toString(), name);
+    }
+
+    private void tagAddedForPayment(final UUID objectId, final String name) {
+        invoicePaymentTagSqlDao.addTag(objectId.toString(), name);
+    }
+
+    private void tagRemovedForPayment(final UUID objectId, final String name) {
+        invoicePaymentTagSqlDao.removeTag(objectId.toString(), name);
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/AnalyticsDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/AnalyticsDao.java
new file mode 100644
index 0000000..8dcc6c8
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/AnalyticsDao.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessAccountTag;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
+
+public interface AnalyticsDao {
+    BusinessAccount getAccountByKey(final String accountKey);
+
+    List<BusinessSubscriptionTransition> getTransitionsByKey(final String externalKey);
+
+    List<BusinessInvoice> getInvoicesByKey(final String accountKey);
+
+    List<BusinessAccountTag> getTagsForAccount(final String accountKey);
+
+    List<BusinessInvoiceItem> getInvoiceItemsForInvoice(final String invoiceId);
+
+    void createInvoice(final String accountKey, final BusinessInvoice invoice, final Iterable<BusinessInvoiceItem> invoiceItems);
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.java
new file mode 100644
index 0000000..d982772
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessAccountField;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessAccountFieldMapper.class)
+public interface BusinessAccountFieldSqlDao {
+    @SqlQuery
+    List<BusinessAccountField> getFieldsForAccount(@Bind("account_key") final String accountKey);
+
+    @SqlUpdate
+    int addField(@Bind("account_key") final String accountKey, @Bind("name") final String name, @Bind("value") final String value);
+
+    @SqlUpdate
+    int removeField(@Bind("account_key") final String accountKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountMapper.java
index a28864b..bf4da3d 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountMapper.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountMapper.java
@@ -34,7 +34,7 @@ public class BusinessAccountMapper implements ResultSetMapper<BusinessAccount> {
                 r.getString(1),
                 r.getString(5),
                 BigDecimal.valueOf(r.getDouble(4)),
-                new DateTime(r.getLong(6), DateTimeZone.UTC),
+                r.getLong(6) == 0 ? null : new DateTime(r.getLong(6), DateTimeZone.UTC),
                 BigDecimal.valueOf(r.getDouble(7)),
                 r.getString(8),
                 r.getString(9),

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountSqlDao.java
index 09a5506..605e604 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountSqlDao.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountSqlDao.java
@@ -20,13 +20,15 @@ import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
 import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
+import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
 import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
 
 import com.ning.billing.analytics.model.BusinessAccount;
 
 @ExternalizedSqlViaStringTemplate3()
 @RegisterMapper(BusinessAccountMapper.class)
-public interface BusinessAccountSqlDao {
+public interface BusinessAccountSqlDao extends Transactional<BusinessAccountSqlDao>, Transmogrifier {
     @SqlQuery
     BusinessAccount getAccount(@Bind("account_key") final String key);
 

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.java
new file mode 100644
index 0000000..0aac244
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessAccountTag;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessAccountTagMapper.class)
+public interface BusinessAccountTagSqlDao {
+    @SqlQuery
+    List<BusinessAccountTag> getTagsForAccount(@Bind("account_key") final String accountKey);
+
+    @SqlUpdate
+    int addTag(@Bind("account_key") final String accountKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    int removeTag(@Bind("account_key") final String accountKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceBinder.java
index ad63e3f..412f323 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceBinder.java
@@ -59,13 +59,13 @@ public @interface BusinessInvoiceBinder {
                     q.bind("account_key", invoice.getAccountKey());
 
                     if (invoice.getInvoiceDate() != null) {
-                        q.bind("invoice_date", invoice.getInvoiceDate());
+                        q.bind("invoice_date", invoice.getInvoiceDate().getMillis());
                     } else {
                         q.bindNull("invoice_date", Types.BIGINT);
                     }
 
                     if (invoice.getTargetDate() != null) {
-                        q.bind("target_date", invoice.getTargetDate());
+                        q.bind("target_date", invoice.getTargetDate().getMillis());
                     } else {
                         q.bindNull("target_date", Types.BIGINT);
                     }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.java
new file mode 100644
index 0000000..a6a9e18
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoiceField;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoiceFieldMapper.class)
+public interface BusinessInvoiceFieldSqlDao {
+    @SqlQuery
+    List<BusinessInvoiceField> getFieldsForInvoice(@Bind("invoice_id") final String invoiceId);
+
+    @SqlUpdate
+    int addField(@Bind("invoice_id") final String invoiceId, @Bind("name") final String name, @Bind("value") final String value);
+
+    @SqlUpdate
+    int removeField(@Bind("invoice_id") final String invoiceId, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemBinder.java
index b171a5c..7ded40e 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemBinder.java
@@ -67,13 +67,13 @@ public @interface BusinessInvoiceItemBinder {
                     q.bind("billing_period", invoiceItem.getBillingPeriod());
 
                     if (invoiceItem.getStartDate() != null) {
-                        q.bind("start_date", invoiceItem.getStartDate());
+                        q.bind("start_date", invoiceItem.getStartDate().getMillis());
                     } else {
                         q.bindNull("start_date", Types.BIGINT);
                     }
 
                     if (invoiceItem.getEndDate() != null) {
-                        q.bind("end_date", invoiceItem.getEndDate());
+                        q.bind("end_date", invoiceItem.getEndDate().getMillis());
                     } else {
                         q.bindNull("end_date", Types.BIGINT);
                     }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.java
index 33c99f7..dd18e3a 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.java
@@ -16,5 +16,39 @@
 
 package com.ning.billing.analytics.dao;
 
-public interface BusinessInvoiceItemSqlDao {
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
+import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoiceItemMapper.class)
+public interface BusinessInvoiceItemSqlDao extends Transactional<BusinessInvoiceItemSqlDao>, Transmogrifier {
+    @SqlQuery
+    BusinessInvoiceItem getInvoiceItem(@Bind("item_id") final String itemId);
+
+    @SqlQuery
+    List<BusinessInvoiceItem> getInvoiceItemsForInvoice(@Bind("invoice_id") final String invoiceId);
+
+    @SqlQuery
+    List<BusinessInvoiceItem> getInvoiceItemsForBundle(@Bind("external_key") final String externalKey);
+
+    @SqlUpdate
+    int createInvoiceItem(@BusinessInvoiceItemBinder final BusinessInvoiceItem invoiceItem);
+
+    @SqlUpdate
+    int updateInvoiceItem(@BusinessInvoiceItemBinder final BusinessInvoiceItem invoiceItem);
+
+    @SqlUpdate
+    int deleteInvoiceItem(@Bind("item_id") final String itemId);
+
+    @SqlUpdate
+    void test();
 }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentBinder.java
index 4cc614b..8c01e8f 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentBinder.java
@@ -61,7 +61,7 @@ public @interface BusinessInvoicePaymentBinder {
                     q.bind("invoice_id", invoicePayment.getInvoiceId().toString());
 
                     if (invoicePayment.getEffectiveDate() != null) {
-                        q.bind("effective_date", invoicePayment.getEffectiveDate());
+                        q.bind("effective_date", invoicePayment.getEffectiveDate().getMillis());
                     } else {
                         q.bindNull("effective_date", Types.BIGINT);
                     }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.java
new file mode 100644
index 0000000..2b85250
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoicePaymentField;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoicePaymentFieldMapper.class)
+public interface BusinessInvoicePaymentFieldSqlDao {
+    @SqlQuery
+    List<BusinessInvoicePaymentField> getFieldsForInvoicePayment(@Bind("payment_id") final String paymentId);
+
+    @SqlUpdate
+    int addField(@Bind("payment_id") final String paymentId, @Bind("name") final String name, @Bind("value") final String value);
+
+    @SqlUpdate
+    int removeField(@Bind("payment_id") final String paymentId, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java
index b825ede..681378d 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java
@@ -16,5 +16,37 @@
 
 package com.ning.billing.analytics.dao;
 
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoicePayment;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoicePaymentMapper.class)
 public interface BusinessInvoicePaymentSqlDao {
+    @SqlQuery
+    BusinessInvoicePayment getInvoicePaymentForPaymentAttempt(@Bind("attempt_id") final String attemptId);
+
+    @SqlQuery
+    List<BusinessInvoicePayment> getInvoicePaymentsForPayment(@Bind("payment_id") final String paymentId);
+
+    @SqlQuery
+    List<BusinessInvoicePayment> getInvoicePaymentsForAccount(@Bind("account_key") final String accountKey);
+
+    @SqlUpdate
+    int createInvoicePayment(@BusinessInvoicePaymentBinder final BusinessInvoicePayment payment);
+
+    @SqlUpdate
+    int updateInvoicePaymentForPaymentAttempt(@BusinessInvoicePaymentBinder final BusinessInvoicePayment payment);
+
+    @SqlUpdate
+    int deleteInvoicePaymentForPaymentAttempt(@Bind("attempt_id") final String attemptId);
+
+    @SqlUpdate
+    void test();
 }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.java
new file mode 100644
index 0000000..74f67b3
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoicePaymentTag;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoicePaymentTagMapper.class)
+public interface BusinessInvoicePaymentTagSqlDao {
+    @SqlQuery
+    List<BusinessInvoicePaymentTag> getTagsForInvoicePayment(@Bind("payment_id") final String paymentId);
+
+    @SqlUpdate
+    int addTag(@Bind("payment_id") final String paymentId, @Bind("name") final String name);
+
+    @SqlUpdate
+    int removeTag(@Bind("payment_id") final String paymentId, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.java
index 2fa5d94..d3c8c6e 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.java
@@ -16,31 +16,35 @@
 
 package com.ning.billing.analytics.dao;
 
+import java.util.List;
+
 import org.skife.jdbi.v2.sqlobject.Bind;
 import org.skife.jdbi.v2.sqlobject.SqlQuery;
 import org.skife.jdbi.v2.sqlobject.SqlUpdate;
 import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
+import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
 import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
 
-import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessInvoice;
 
 @ExternalizedSqlViaStringTemplate3()
-@RegisterMapper(BusinessAccountMapper.class)
-public interface BusinessInvoiceSqlDao {
+@RegisterMapper(BusinessInvoiceMapper.class)
+public interface BusinessInvoiceSqlDao extends Transactional<BusinessInvoiceSqlDao>, Transmogrifier {
     @SqlQuery
-    BusinessAccount getInvoice(@Bind("invoice_id") final String invoiceId);
+    BusinessInvoice getInvoice(@Bind("invoice_id") final String invoiceId);
 
     @SqlQuery
-    BusinessAccount getInvoicesForAccount(@Bind("account_key") final String accountKey);
+    List<BusinessInvoice> getInvoicesForAccount(@Bind("account_key") final String accountKey);
 
     @SqlUpdate
-    int createInvoice(final BusinessAccount account);
+    int createInvoice(@BusinessInvoiceBinder final BusinessInvoice invoice);
 
     @SqlUpdate
-    int updateInvoice(@BusinessAccountBinder final BusinessAccount account);
+    int updateInvoice(@BusinessInvoiceBinder final BusinessInvoice invoice);
 
     @SqlUpdate
-    int deleteInvoice(@BusinessAccountBinder final BusinessAccount account);
+    int deleteInvoice(@Bind("invoice_id") final String invoiceId);
 
     @SqlUpdate
     void test();

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.java
new file mode 100644
index 0000000..587f45a
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessInvoiceTag;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoiceTagMapper.class)
+public interface BusinessInvoiceTagSqlDao {
+    @SqlQuery
+    List<BusinessInvoiceTag> getTagsForInvoice(@Bind("invoice_id") final String invoiceId);
+
+    @SqlUpdate
+    int addTag(@Bind("invoice_id") final String invoiceId, @Bind("name") final String name);
+
+    @SqlUpdate
+    int removeTag(@Bind("invoice_id") final String invoiceId, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusBinder.java
index de9629a..eb1bd9d 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusBinder.java
@@ -42,13 +42,13 @@ public @interface BusinessOverdueStatusBinder {
                     q.bind("status", overdueStatus.getStatus());
 
                     if (overdueStatus.getStartDate() != null) {
-                        q.bind("start_date", overdueStatus.getStartDate());
+                        q.bind("start_date", overdueStatus.getStartDate().getMillis());
                     } else {
                         q.bindNull("start_date", Types.BIGINT);
                     }
 
                     if (overdueStatus.getEndDate() != null) {
-                        q.bind("end_date", overdueStatus.getEndDate());
+                        q.bind("end_date", overdueStatus.getEndDate().getMillis());
                     } else {
                         q.bindNull("end_date", Types.BIGINT);
                     }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.java
new file mode 100644
index 0000000..44668d0
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessOverdueStatus;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessOverdueStatusMapper.class)
+public interface BusinessOverdueStatusSqlDao {
+    @SqlQuery
+    List<BusinessOverdueStatus> getOverdueStatusesForBundle(@Bind("external_key") final String externalKey);
+
+    @SqlUpdate
+    int createOverdueStatus(@BusinessOverdueStatusBinder final BusinessOverdueStatus status);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.java
new file mode 100644
index 0000000..d22076e
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionField;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessSubscriptionTransitionFieldMapper.class)
+public interface BusinessSubscriptionTransitionFieldSqlDao {
+    @SqlQuery
+    List<BusinessSubscriptionTransitionField> getFieldsForBusinessSubscriptionTransition(@Bind("external_key") final String externalKey);
+
+    @SqlUpdate
+    int addField(@Bind("external_key") final String externalKey, @Bind("name") final String name, @Bind("value") final String value);
+
+    @SqlUpdate
+    int removeField(@Bind("external_key") final String externalKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.java
new file mode 100644
index 0000000..c073a53
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+
+import org.skife.jdbi.v2.sqlobject.Bind;
+import org.skife.jdbi.v2.sqlobject.SqlQuery;
+import org.skife.jdbi.v2.sqlobject.SqlUpdate;
+import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper;
+import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
+
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionTag;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessSubscriptionTransitionTagMapper.class)
+public interface BusinessSubscriptionTransitionTagSqlDao {
+    @SqlQuery
+    List<BusinessSubscriptionTransitionTag> getTagsForBusinessSubscriptionTransition(@Bind("external_key") final String externalKey);
+
+    @SqlUpdate
+    int addTag(@Bind("external_key") final String externalKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    int removeTag(@Bind("external_key") final String externalKey, @Bind("name") final String name);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/DefaultAnalyticsDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/DefaultAnalyticsDao.java
new file mode 100644
index 0000000..d5f8627
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/DefaultAnalyticsDao.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import javax.inject.Inject;
+import java.util.List;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.Transaction;
+import org.skife.jdbi.v2.TransactionStatus;
+
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessAccountTag;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
+
+public class DefaultAnalyticsDao implements AnalyticsDao {
+    private final BusinessAccountSqlDao accountSqlDao;
+    private final BusinessSubscriptionTransitionSqlDao subscriptionTransitionSqlDao;
+    private final BusinessInvoiceSqlDao invoiceSqlDao;
+    private final BusinessInvoiceItemSqlDao invoiceItemSqlDao;
+    private final BusinessAccountTagSqlDao accountTagSqlDao;
+
+    @Inject
+    public DefaultAnalyticsDao(final BusinessAccountSqlDao accountSqlDao,
+                               final BusinessSubscriptionTransitionSqlDao subscriptionTransitionSqlDao,
+                               final BusinessInvoiceSqlDao invoiceSqlDao,
+                               final BusinessInvoiceItemSqlDao invoiceItemSqlDao,
+                               final BusinessAccountTagSqlDao accountTagSqlDao) {
+        this.accountSqlDao = accountSqlDao;
+        this.subscriptionTransitionSqlDao = subscriptionTransitionSqlDao;
+        this.invoiceSqlDao = invoiceSqlDao;
+        this.invoiceItemSqlDao = invoiceItemSqlDao;
+        this.accountTagSqlDao = accountTagSqlDao;
+    }
+
+    @Override
+    public BusinessAccount getAccountByKey(final String accountKey) {
+        return accountSqlDao.getAccount(accountKey);
+    }
+
+    @Override
+    public List<BusinessSubscriptionTransition> getTransitionsByKey(final String externalKey) {
+        return subscriptionTransitionSqlDao.getTransitions(externalKey);
+    }
+
+    @Override
+    public List<BusinessInvoice> getInvoicesByKey(final String accountKey) {
+        return invoiceSqlDao.getInvoicesForAccount(accountKey);
+    }
+
+    @Override
+    public List<BusinessAccountTag> getTagsForAccount(final String accountKey) {
+        return accountTagSqlDao.getTagsForAccount(accountKey);
+    }
+
+    @Override
+    public List<BusinessInvoiceItem> getInvoiceItemsForInvoice(final String invoiceId) {
+        return invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceId);
+    }
+
+    @Override
+    public void createInvoice(final String accountKey, final BusinessInvoice invoice, final Iterable<BusinessInvoiceItem> invoiceItems) {
+        invoiceSqlDao.inTransaction(new Transaction<Void, BusinessInvoiceSqlDao>() {
+            @Override
+            public Void inTransaction(final BusinessInvoiceSqlDao transactional, final TransactionStatus status) throws Exception {
+                // Create the invoice
+                transactional.createInvoice(invoice);
+
+                // Add associated invoice items
+                final BusinessInvoiceItemSqlDao invoiceItemSqlDao = transactional.become(BusinessInvoiceItemSqlDao.class);
+                for (final BusinessInvoiceItem invoiceItem : invoiceItems) {
+                    invoiceItemSqlDao.createInvoiceItem(invoiceItem);
+                }
+
+                // Update BAC
+                final BusinessAccountSqlDao accountSqlDao = transactional.become(BusinessAccountSqlDao.class);
+                final BusinessAccount account = accountSqlDao.getAccount(accountKey);
+                if (account == null) {
+                    throw new IllegalStateException("Account does not exist for key " + accountKey);
+                }
+                account.setBalance(account.getBalance().add(invoice.getBalance()));
+                account.setLastInvoiceDate(invoice.getInvoiceDate());
+                account.setTotalInvoiceBalance(account.getTotalInvoiceBalance().add(invoice.getBalance()));
+                account.setUpdatedDt(new DateTime(DateTimeZone.UTC));
+                accountSqlDao.saveAccount(account);
+
+                return null;
+            }
+        });
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccount.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccount.java
index a74fea3..b8b056c 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccount.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccount.java
@@ -27,7 +27,7 @@ public class BusinessAccount {
     private DateTime createdDt = null;
     private DateTime updatedDt = null;
 
-    private final String key;
+    private String key;
     private String name;
     private BigDecimal balance;
     private DateTime lastInvoiceDate;
@@ -37,6 +37,9 @@ public class BusinessAccount {
     private String creditCardType;
     private String billingAddressCountry;
 
+    public BusinessAccount() {
+    }
+
     public BusinessAccount(final String key, final String name, final BigDecimal balance, final DateTime lastInvoiceDate, final BigDecimal totalInvoiceBalance, final String lastPaymentStatus, final String paymentMethod, final String creditCardType, final String billingAddressCountry) {
         this.key = key;
         this.balance = balance;
@@ -53,6 +56,10 @@ public class BusinessAccount {
         return key;
     }
 
+    public void setKey(final String key) {
+        this.key = key;
+    }
+
     public BigDecimal getBalance() {
         return balance;
     }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoice.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoice.java
index 2c6e8bc..31591c6 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoice.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoice.java
@@ -20,8 +20,11 @@ import java.math.BigDecimal;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
 
+import com.ning.billing.analytics.utils.Rounder;
 import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.invoice.api.Invoice;
 
 public class BusinessInvoice {
     private final UUID invoiceId;
@@ -54,6 +57,12 @@ public class BusinessInvoice {
         this.updatedDate = updatedDate;
     }
 
+    public BusinessInvoice(final String accountKey, final Invoice invoice) {
+        this(accountKey, invoice.getAmountCharged(), invoice.getAmountCredited(), invoice.getAmountPaid(), invoice.getBalance(),
+             new DateTime(DateTimeZone.UTC), invoice.getCurrency(), invoice.getInvoiceDate(), invoice.getId(), invoice.getTargetDate(),
+             new DateTime(DateTimeZone.UTC));
+    }
+
     public DateTime getCreatedDate() {
         return createdDate;
     }
@@ -167,16 +176,16 @@ public class BusinessInvoice {
         if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
             return false;
         }
-        if (amountCharged != null ? !amountCharged.equals(that.amountCharged) : that.amountCharged != null) {
+        if (amountCharged != null ? Rounder.round(amountCharged) != Rounder.round(that.amountCharged) : that.amountCharged != null) {
             return false;
         }
-        if (amountCredited != null ? !amountCredited.equals(that.amountCredited) : that.amountCredited != null) {
+        if (amountCredited != null ? Rounder.round(amountCredited) != Rounder.round(that.amountCredited) : that.amountCredited != null) {
             return false;
         }
-        if (amountPaid != null ? !amountPaid.equals(that.amountPaid) : that.amountPaid != null) {
+        if (amountPaid != null ? Rounder.round(amountPaid) != Rounder.round(that.amountPaid) : that.amountPaid != null) {
             return false;
         }
-        if (balance != null ? !balance.equals(that.balance) : that.balance != null) {
+        if (balance != null ? Rounder.round(balance) != Rounder.round(that.balance) : that.balance != null) {
             return false;
         }
         if (createdDate != null ? !createdDate.equals(that.createdDate) : that.createdDate != null) {

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceItem.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceItem.java
index eb2793f..03cb50c 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceItem.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceItem.java
@@ -20,8 +20,13 @@ import java.math.BigDecimal;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
 
+import com.ning.billing.analytics.utils.Rounder;
 import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.catalog.api.Plan;
+import com.ning.billing.catalog.api.PlanPhase;
+import com.ning.billing.invoice.api.InvoiceItem;
 
 public class BusinessInvoiceItem {
     private final UUID itemId;
@@ -65,6 +70,13 @@ public class BusinessInvoiceItem {
         this.updatedDate = updatedDate;
     }
 
+    public BusinessInvoiceItem(final String externalKey, final InvoiceItem invoiceItem, final Plan plan, final PlanPhase planPhase) {
+        this(invoiceItem.getAmount(), planPhase.getBillingPeriod().toString(), new DateTime(DateTimeZone.UTC), invoiceItem.getCurrency(), invoiceItem.getEndDate(),
+             externalKey, invoiceItem.getInvoiceId(), invoiceItem.getId(), invoiceItem.getInvoiceItemType().toString(),
+             planPhase.getPhaseType().toString(), plan.getProduct().getCategory().toString(), plan.getProduct().getName(), plan.getProduct().getCatalogName(),
+             planPhase.getName(), invoiceItem.getStartDate(), new DateTime(DateTimeZone.UTC));
+    }
+
     public DateTime getCreatedDate() {
         return createdDate;
     }
@@ -220,7 +232,7 @@ public class BusinessInvoiceItem {
 
         final BusinessInvoiceItem that = (BusinessInvoiceItem) o;
 
-        if (amount != null ? !amount.equals(that.amount) : that.amount != null) {
+        if (amount != null ? Rounder.round(amount) != (Rounder.round(that.amount)) : that.amount != null) {
             return false;
         }
         if (billingPeriod != null ? !billingPeriod.equals(that.billingPeriod) : that.billingPeriod != null) {

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePayment.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePayment.java
index a0ea526..2622437 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePayment.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePayment.java
@@ -21,6 +21,7 @@ import java.util.UUID;
 
 import org.joda.time.DateTime;
 
+import com.ning.billing.analytics.utils.Rounder;
 import com.ning.billing.catalog.api.Currency;
 
 public class BusinessInvoicePayment {
@@ -231,7 +232,7 @@ public class BusinessInvoicePayment {
         if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
             return false;
         }
-        if (amount != null ? !amount.equals(that.amount) : that.amount != null) {
+        if (amount != null ? Rounder.round(amount) != Rounder.round(that.amount) : that.amount != null) {
             return false;
         }
         if (attemptId != null ? !attemptId.equals(that.attemptId) : that.attemptId != null) {
@@ -273,7 +274,7 @@ public class BusinessInvoicePayment {
         if (processingStatus != null ? !processingStatus.equals(that.processingStatus) : that.processingStatus != null) {
             return false;
         }
-        if (requestedAmount != null ? !requestedAmount.equals(that.requestedAmount) : that.requestedAmount != null) {
+        if (requestedAmount != null ? Rounder.round(requestedAmount) != Rounder.round(that.requestedAmount) : that.requestedAmount != null) {
             return false;
         }
         if (updatedDate != null ? !updatedDate.equals(that.updatedDate) : that.updatedDate != null) {

diff --git a/analytics/src/main/java/com/ning/billing/analytics/setup/AnalyticsModule.java b/analytics/src/main/java/com/ning/billing/analytics/setup/AnalyticsModule.java
index bb564e6..82433ec 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/setup/AnalyticsModule.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/setup/AnalyticsModule.java
@@ -21,23 +21,52 @@ import com.google.inject.AbstractModule;
 import com.ning.billing.analytics.AnalyticsListener;
 import com.ning.billing.analytics.BusinessAccountRecorder;
 import com.ning.billing.analytics.BusinessSubscriptionTransitionRecorder;
+import com.ning.billing.analytics.BusinessTagRecorder;
 import com.ning.billing.analytics.api.AnalyticsService;
 import com.ning.billing.analytics.api.DefaultAnalyticsService;
+import com.ning.billing.analytics.api.user.DefaultAnalyticsUserApi;
+import com.ning.billing.analytics.dao.AnalyticsDao;
 import com.ning.billing.analytics.dao.BusinessAccountSqlDao;
-import com.ning.billing.analytics.dao.BusinessAccountSqlDaoProvider;
+import com.ning.billing.analytics.dao.BusinessAccountTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceFieldSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceItemSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoicePaymentFieldSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoicePaymentSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoicePaymentTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessOverdueStatusSqlDao;
+import com.ning.billing.analytics.dao.BusinessSqlProvider;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionFieldSqlDao;
 import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDao;
-import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDaoProvider;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionTagSqlDao;
+import com.ning.billing.analytics.dao.DefaultAnalyticsDao;
 
 public class AnalyticsModule extends AbstractModule {
     @Override
     protected void configure() {
-        bind(BusinessSubscriptionTransitionSqlDao.class).toProvider(BusinessSubscriptionTransitionSqlDaoProvider.class).asEagerSingleton();
-        bind(BusinessAccountSqlDao.class).toProvider(BusinessAccountSqlDaoProvider.class).asEagerSingleton();
+        bind(BusinessAccountSqlDao.class).toProvider(new BusinessSqlProvider<BusinessAccountSqlDao>(BusinessAccountSqlDao.class));
+        bind(BusinessAccountTagSqlDao.class).toProvider(new BusinessSqlProvider<BusinessAccountTagSqlDao>(BusinessAccountTagSqlDao.class));
+        bind(BusinessInvoiceFieldSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoiceFieldSqlDao>(BusinessInvoiceFieldSqlDao.class));
+        bind(BusinessInvoiceItemSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoiceItemSqlDao>(BusinessInvoiceItemSqlDao.class));
+        bind(BusinessInvoicePaymentFieldSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoicePaymentFieldSqlDao>(BusinessInvoicePaymentFieldSqlDao.class));
+        bind(BusinessInvoicePaymentSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoicePaymentSqlDao>(BusinessInvoicePaymentSqlDao.class));
+        bind(BusinessInvoicePaymentTagSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoicePaymentTagSqlDao>(BusinessInvoicePaymentTagSqlDao.class));
+        bind(BusinessInvoiceSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoiceSqlDao>(BusinessInvoiceSqlDao.class));
+        bind(BusinessInvoiceTagSqlDao.class).toProvider(new BusinessSqlProvider<BusinessInvoiceTagSqlDao>(BusinessInvoiceTagSqlDao.class));
+        bind(BusinessOverdueStatusSqlDao.class).toProvider(new BusinessSqlProvider<BusinessOverdueStatusSqlDao>(BusinessOverdueStatusSqlDao.class));
+        bind(BusinessSubscriptionTransitionFieldSqlDao.class).toProvider(new BusinessSqlProvider<BusinessSubscriptionTransitionFieldSqlDao>(BusinessSubscriptionTransitionFieldSqlDao.class));
+        bind(BusinessSubscriptionTransitionSqlDao.class).toProvider(new BusinessSqlProvider<BusinessSubscriptionTransitionSqlDao>(BusinessSubscriptionTransitionSqlDao.class));
+        bind(BusinessSubscriptionTransitionTagSqlDao.class).toProvider(new BusinessSqlProvider<BusinessSubscriptionTransitionTagSqlDao>(BusinessSubscriptionTransitionTagSqlDao.class));
 
         bind(BusinessSubscriptionTransitionRecorder.class).asEagerSingleton();
         bind(BusinessAccountRecorder.class).asEagerSingleton();
+        bind(BusinessTagRecorder.class).asEagerSingleton();
         bind(AnalyticsListener.class).asEagerSingleton();
 
+        bind(AnalyticsDao.class).to(DefaultAnalyticsDao.class).asEagerSingleton();
         bind(AnalyticsService.class).to(DefaultAnalyticsService.class).asEagerSingleton();
+
+        bind(DefaultAnalyticsUserApi.class).asEagerSingleton();
     }
 }

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.sql.stg
new file mode 100644
index 0000000..8bb9e0b
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountFieldSqlDao.sql.stg
@@ -0,0 +1,31 @@
+group BusinessAccountField;
+
+getFieldsForAccount(account_key) ::=<<
+select
+  account_key
+, name
+, value
+from bac_fields
+where account_key = :account_key
+;
+>>
+
+addField(account_key, name, value) ::=<<
+insert into bac_fields (
+  account_key
+, name
+, value
+) values (
+  :account_key
+, :name
+, :value
+);
+>>
+
+removeField(account_key, name) ::= <<
+delete from bac_fields where account_key = :account_key and name = :name;
+>>
+
+test() ::= <<
+select 1 from bac_fields;
+>>
\ No newline at end of file

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.sql.stg
new file mode 100644
index 0000000..af34473
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessAccountTagSqlDao.sql.stg
@@ -0,0 +1,28 @@
+group BusinessAccountTag;
+
+getTagsForAccount(account_key) ::=<<
+select
+  account_key
+, name
+from bac_tags
+where account_key = :account_key
+;
+>>
+
+addTag(account_key, name) ::=<<
+insert into bac_tags (
+  account_key
+, name
+) values (
+  :account_key
+, :name
+);
+>>
+
+removeTag(account_key, name) ::= <<
+delete from bac_tags where account_key = :account_key and name = :name;
+>>
+
+test() ::= <<
+select 1 from bac_tags;
+>>
\ No newline at end of file

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.sql.stg
new file mode 100644
index 0000000..fe0080f
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.sql.stg
@@ -0,0 +1,31 @@
+group BusinessInvoiceField;
+
+getFieldsForInvoice(invoice_id) ::=<<
+select
+  invoice_id
+, name
+, value
+from bin_fields
+where invoice_id = :invoice_id
+;
+>>
+
+addField(invoice_id, name, value) ::=<<
+insert into bin_fields (
+  invoice_id
+, name
+, value
+) values (
+  :invoice_id
+, :name
+, :value
+);
+>>
+
+removeField(invoice_id, name, value) ::= <<
+delete from bin_fields where invoice_id = :invoice_id and name = :name;
+>>
+
+test() ::= <<
+select 1 from bin_tags;
+>>

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.sql.stg
index 44862ea..7952eb6 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.sql.stg
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.sql.stg
@@ -44,6 +44,7 @@ select
 , currency
 from bii
 where invoice_id = :invoice_id
+order by created_date asc
 ;
 >>
 
@@ -67,6 +68,7 @@ select
 , currency
 from bii
 where external_key = :external_key
+order by created_date asc
 ;
 >>
 

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.sql.stg
new file mode 100644
index 0000000..a2ced92
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldSqlDao.sql.stg
@@ -0,0 +1,31 @@
+group BusinessInvoicePaymentField;
+
+getFieldsForInvoicePayment(payment_id) ::=<<
+select
+  payment_id
+, name
+, value
+from bip_fields
+where payment_id = :payment_id
+;
+>>
+
+addField(payment_id, name, value) ::=<<
+insert into bip_fields (
+  payment_id
+, name
+, value
+) values (
+  :payment_id
+, :name
+, :value
+);
+>>
+
+removeField(payment_id, name) ::= <<
+delete from bip_fields where payment_id = :payment_id and name = :name;
+>>
+
+test() ::= <<
+select 1 from bip_fields;
+>>

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.sql.stg
index 402d500..c5d0fb8 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.sql.stg
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.sql.stg
@@ -46,6 +46,7 @@ select
 , card_country
 from bip
 where payment_id = :payment_id
+order by created_date asc
 ;
 >>
 
@@ -70,6 +71,7 @@ select
 , card_country
 from bip
 where account_key = :account_key
+order by created_date asc
 ;
 >>
 
@@ -113,7 +115,7 @@ insert into bip (
 );
 >>
 
-updateInvoicePaymentforPaymentAttempt(attempt_id) ::= <<
+updateInvoicePaymentForPaymentAttempt() ::= <<
 update bip set
   updated_date = :updated_date
 , account_key = :account_key

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.sql.stg
new file mode 100644
index 0000000..57553ad
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagSqlDao.sql.stg
@@ -0,0 +1,28 @@
+group BusinessInvoicePaymentTag;
+
+getTagsForInvoicePayment(payment_id) ::=<<
+select
+  payment_id
+, name
+from bip_tags
+where payment_id = :payment_id
+;
+>>
+
+addTag(payment_id, name) ::=<<
+insert into bip_tags (
+  payment_id
+, name
+) values (
+  :payment_id
+, :name
+);
+>>
+
+removeTag(payment_id, name) ::= <<
+delete from bip_tags where payment_id = :payment_id and name = :name;
+>>
+
+test() ::= <<
+select 1 from bip_tags;
+>>

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.sql.stg
index 7d72ef0..83d3ffc 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.sql.stg
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.sql.stg
@@ -34,6 +34,7 @@ select
 , amount_credited
 from bin
 where account_key = :account_key
+order by created_date asc
 ;
 >>
 

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.sql.stg
new file mode 100644
index 0000000..f01a7d7
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceTagSqlDao.sql.stg
@@ -0,0 +1,28 @@
+group BusinessInvoiceTag;
+
+getTagsForInvoice(invoice_id) ::=<<
+select
+  invoice_id
+, name
+from bin_tags
+where invoice_id = :invoice_id
+;
+>>
+
+addTag(invoice_id, name) ::=<<
+insert into bin_tags (
+  invoice_id
+, name
+) values (
+  :invoice_id
+, :name
+);
+>>
+
+removeTag(invoice_id, name) ::= <<
+delete from bin_tags where invoice_id = :invoice_id and name = :name;
+>>
+
+test() ::= <<
+select 1 from bin_tags;
+>>

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.sql.stg
index 7153cc2..d412125 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.sql.stg
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.sql.stg
@@ -8,6 +8,7 @@ select
 , end_date
 from bos
 where external_key = :external_key
+order by start_date asc
 ;
 >>
 

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.sql.stg
new file mode 100644
index 0000000..522e6d7
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldSqlDao.sql.stg
@@ -0,0 +1,31 @@
+group BusinessSubscriptionTransitionField;
+
+getFieldsForBusinessSubscriptionTransition(external_key) ::=<<
+select
+  external_key
+, name
+, value
+from bst_fields
+where external_key = :external_key
+;
+>>
+
+addField(external_key, name, value) ::=<<
+insert into bst_fields (
+  external_key
+, name
+, value
+) values (
+  :external_key
+, :name
+, :value
+);
+>>
+
+removeField(external_key, name) ::= <<
+delete from bst_fields where external_key = :external_key and name = :name;
+>>
+
+test() ::= <<
+select 1 from bst_fields;
+>>
\ No newline at end of file

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.sql.stg b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.sql.stg
new file mode 100644
index 0000000..3d2255f
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagSqlDao.sql.stg
@@ -0,0 +1,28 @@
+group BusinessSubscriptionTransitionTag;
+
+getTagsForBusinessSubscriptionTransition(external_key) ::=<<
+select
+  external_key
+, name
+from bst_tags
+where external_key = :external_key
+;
+>>
+
+addTag(external_key, name) ::=<<
+insert into bst_tags (
+  external_key
+, name
+) values (
+  :external_key
+, :name
+);
+>>
+
+removeTag(external_key, name) ::= <<
+delete from bst_tags where external_key = :external_key and name = :name;
+>>
+
+test() ::= <<
+select 1 from bst_tags;
+>>
\ No newline at end of file

diff --git a/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql b/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql
index 1f27d07..43aaad8 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql
+++ b/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql
@@ -34,7 +34,7 @@ create table bst (
 , next_subscription_id varchar(100) default null
 , next_bundle_id varchar(100) default null
 , primary key(total_ordering)
-) engine=innodb comment 'Business Subscription Transitions, track bundles';
+) engine=innodb comment 'Business Subscription Transitions, track bundles lifecycle';
 create index bst_key_index on bst (external_key, requested_timestamp asc);
 
 drop table if exists bac;

diff --git a/analytics/src/test/java/com/ning/billing/analytics/api/TestAnalyticsService.java b/analytics/src/test/java/com/ning/billing/analytics/api/TestAnalyticsService.java
index 23209b9..82b1241 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/api/TestAnalyticsService.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/api/TestAnalyticsService.java
@@ -241,15 +241,8 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
     */
     }
 
-
-    @AfterClass(groups = "slow")
-    public void stopMysql() {
-        helper.stopMysql();
-    }
-
-
     // STEPH talk to Pierre -- see previous remark hence disable test
-    @Test(groups = "slow", enabled = true)
+    @Test(groups = "slow", enabled = false)
     public void testRegisterForNotifications() throws Exception {
         // Make sure the service has been instantiated
         Assert.assertEquals(service.getName(), "analytics-service");

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountFieldSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountFieldSqlDao.java
new file mode 100644
index 0000000..69fed56
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountFieldSqlDao.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessAccountField;
+
+public class TestBusinessAccountFieldSqlDao extends TestWithEmbeddedDB {
+    private BusinessAccountFieldSqlDao accountFieldSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        accountFieldSqlDao = dbi.onDemand(BusinessAccountFieldSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 30);
+        final String value = UUID.randomUUID().toString();
+
+        // Verify initial state
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey).size(), 0);
+        Assert.assertEquals(accountFieldSqlDao.removeField(accountKey, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(accountFieldSqlDao.addField(accountKey, name, value), 1);
+        final List<BusinessAccountField> fieldsForAccount = accountFieldSqlDao.getFieldsForAccount(accountKey);
+        Assert.assertEquals(fieldsForAccount.size(), 1);
+
+        // Retrieve it
+        final BusinessAccountField accountField = fieldsForAccount.get(0);
+        Assert.assertEquals(accountField.getAccountKey(), accountKey);
+        Assert.assertEquals(accountField.getName(), name);
+        Assert.assertEquals(accountField.getValue(), value);
+
+        // Delete it
+        Assert.assertEquals(accountFieldSqlDao.removeField(accountKey, name), 1);
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String accountKey1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 30);
+        final String accountKey2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 30);
+
+        // Add a field to both accounts
+        Assert.assertEquals(accountFieldSqlDao.addField(accountKey1, name1, UUID.randomUUID().toString()), 1);
+        Assert.assertEquals(accountFieldSqlDao.addField(accountKey2, name2, UUID.randomUUID().toString()), 1);
+
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey1).size(), 1);
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey2).size(), 1);
+
+        // Remove the field for the first account
+        Assert.assertEquals(accountFieldSqlDao.removeField(accountKey1, name1), 1);
+
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey1).size(), 0);
+        Assert.assertEquals(accountFieldSqlDao.getFieldsForAccount(accountKey2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            accountFieldSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountTagSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountTagSqlDao.java
new file mode 100644
index 0000000..20ccfc8
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessAccountTagSqlDao.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessAccountTag;
+
+public class TestBusinessAccountTagSqlDao extends TestWithEmbeddedDB {
+    private BusinessAccountTagSqlDao accountTagSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        accountTagSqlDao = dbi.onDemand(BusinessAccountTagSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+
+        // Verify initial state
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey).size(), 0);
+        Assert.assertEquals(accountTagSqlDao.removeTag(accountKey, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(accountTagSqlDao.addTag(accountKey, name), 1);
+        final List<BusinessAccountTag> tagsForAccount = accountTagSqlDao.getTagsForAccount(accountKey);
+        Assert.assertEquals(tagsForAccount.size(), 1);
+
+        // Retrieve it
+        final BusinessAccountTag accountTag = tagsForAccount.get(0);
+        Assert.assertEquals(accountTag.getAccountKey(), accountKey);
+        Assert.assertEquals(accountTag.getName(), name);
+
+        // Delete it
+        Assert.assertEquals(accountTagSqlDao.removeTag(accountKey, name), 1);
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String accountKey1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 20);
+        final String accountKey2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 20);
+
+        // Add a tag to both accounts
+        Assert.assertEquals(accountTagSqlDao.addTag(accountKey1, name1), 1);
+        Assert.assertEquals(accountTagSqlDao.addTag(accountKey2, name2), 1);
+
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey1).size(), 1);
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey2).size(), 1);
+
+        // Remove the tag for the first account
+        Assert.assertEquals(accountTagSqlDao.removeTag(accountKey1, name1), 1);
+
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey1).size(), 0);
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            accountTagSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceFieldSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceFieldSqlDao.java
new file mode 100644
index 0000000..9553dfa
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceFieldSqlDao.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoiceField;
+
+public class TestBusinessInvoiceFieldSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoiceFieldSqlDao invoiceFieldSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoiceFieldSqlDao = dbi.onDemand(BusinessInvoiceFieldSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String invoiceId = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 30);
+        final String value = UUID.randomUUID().toString();
+
+        // Verify initial state
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId).size(), 0);
+        Assert.assertEquals(invoiceFieldSqlDao.removeField(invoiceId, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(invoiceFieldSqlDao.addField(invoiceId, name, value), 1);
+        final List<BusinessInvoiceField> fieldsForInvoice = invoiceFieldSqlDao.getFieldsForInvoice(invoiceId);
+        Assert.assertEquals(fieldsForInvoice.size(), 1);
+
+        // Retrieve it
+        final BusinessInvoiceField invoiceField = fieldsForInvoice.get(0);
+        Assert.assertEquals(invoiceField.getInvoiceId().toString(), invoiceId);
+        Assert.assertEquals(invoiceField.getName(), name);
+        Assert.assertEquals(invoiceField.getValue(), value);
+
+        // Delete it
+        Assert.assertEquals(invoiceFieldSqlDao.removeField(invoiceId, name), 1);
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String invoiceId1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 30);
+        final String invoiceId2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 30);
+
+        // Add a field to both invoices
+        Assert.assertEquals(invoiceFieldSqlDao.addField(invoiceId1, name1, UUID.randomUUID().toString()), 1);
+        Assert.assertEquals(invoiceFieldSqlDao.addField(invoiceId2, name2, UUID.randomUUID().toString()), 1);
+
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId1).size(), 1);
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId2).size(), 1);
+
+        // Remove the field for the first invoice
+        Assert.assertEquals(invoiceFieldSqlDao.removeField(invoiceId1, name1), 1);
+
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId1).size(), 0);
+        Assert.assertEquals(invoiceFieldSqlDao.getFieldsForInvoice(invoiceId2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoiceFieldSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceItemSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceItemSqlDao.java
new file mode 100644
index 0000000..1bd7b31
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceItemSqlDao.java
@@ -0,0 +1,134 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoiceItemSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoiceItemSqlDao invoiceItemSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoiceItemSqlDao = dbi.onDemand(BusinessInvoiceItemSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final UUID invoiceId = UUID.randomUUID();
+        final String externalKey = UUID.randomUUID().toString();
+        final BusinessInvoiceItem invoiceItem = createInvoiceItem(invoiceId, externalKey);
+
+        // Verify initial state
+        Assert.assertNull(invoiceItemSqlDao.getInvoiceItem(invoiceItem.getItemId().toString()));
+        Assert.assertEquals(invoiceItemSqlDao.deleteInvoiceItem(invoiceItem.getItemId().toString()), 0);
+
+        // Add the invoice item
+        Assert.assertEquals(invoiceItemSqlDao.createInvoiceItem(invoiceItem), 1);
+
+        // Retrieve it
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItem(invoiceItem.getItemId().toString()), invoiceItem);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(invoiceItem.getExternalKey()).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(invoiceItem.getExternalKey()).get(0), invoiceItem);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceItem.getInvoiceId().toString()).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceItem.getInvoiceId().toString()).get(0), invoiceItem);
+
+        // Update and retrieve it
+        invoiceItem.setProductName(UUID.randomUUID().toString().substring(0, 20));
+        Assert.assertEquals(invoiceItemSqlDao.updateInvoiceItem(invoiceItem), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItem(invoiceItem.getItemId().toString()), invoiceItem);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(invoiceItem.getExternalKey()).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(invoiceItem.getExternalKey()).get(0), invoiceItem);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceItem.getInvoiceId().toString()).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceItem.getInvoiceId().toString()).get(0), invoiceItem);
+
+        // Delete it
+        Assert.assertEquals(invoiceItemSqlDao.deleteInvoiceItem(invoiceItem.getItemId().toString()), 1);
+        Assert.assertNull(invoiceItemSqlDao.getInvoiceItem(invoiceItem.getItemId().toString()));
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(invoiceItem.getExternalKey()).size(), 0);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceItem.getInvoiceId().toString()).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final UUID invoiceId1 = UUID.randomUUID();
+        final String externalKey1 = UUID.randomUUID().toString();
+        final BusinessInvoiceItem invoiceItem1 = createInvoiceItem(invoiceId1, externalKey1);
+        final UUID invoiceId2 = UUID.randomUUID();
+        final String externalKey2 = UUID.randomUUID().toString();
+        final BusinessInvoiceItem invoiceItem2 = createInvoiceItem(invoiceId2, externalKey2);
+
+        // Create both invoice items
+        Assert.assertEquals(invoiceItemSqlDao.createInvoiceItem(invoiceItem1), 1);
+        Assert.assertEquals(invoiceItemSqlDao.createInvoiceItem(invoiceItem2), 1);
+
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(externalKey1).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(externalKey2).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceId1.toString()).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceId2.toString()).size(), 1);
+
+        // Remove the first invoice item
+        Assert.assertEquals(invoiceItemSqlDao.deleteInvoiceItem(invoiceItem1.getItemId().toString()), 1);
+
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(externalKey1).size(), 0);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForBundle(externalKey2).size(), 1);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceId1.toString()).size(), 0);
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoiceId2.toString()).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoiceItemSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+
+    private BusinessInvoiceItem createInvoiceItem(final UUID invoiceId, final String externalKey) {
+        final BigDecimal amount = BigDecimal.TEN;
+        final String billingPeriod = UUID.randomUUID().toString().substring(0, 20);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.AUD;
+        final DateTime endDate = new DateTime(DateTimeZone.UTC);
+        final UUID itemId = UUID.randomUUID();
+        final String itemType = UUID.randomUUID().toString().substring(0, 20);
+        final String phase = UUID.randomUUID().toString().substring(0, 20);
+        final String productCategory = UUID.randomUUID().toString().substring(0, 20);
+        final String productName = UUID.randomUUID().toString().substring(0, 20);
+        final String productType = UUID.randomUUID().toString().substring(0, 20);
+        final String slug = UUID.randomUUID().toString();
+        final DateTime startDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+
+        return new BusinessInvoiceItem(amount, billingPeriod, createdDate, currency, endDate, externalKey, invoiceId,
+                                       itemId, itemType, phase, productCategory, productName, productType, slug, startDate, updatedDate);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentFieldSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentFieldSqlDao.java
new file mode 100644
index 0000000..b97849d
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentFieldSqlDao.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoicePaymentField;
+
+public class TestBusinessInvoicePaymentFieldSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoicePaymentFieldSqlDao invoicePaymentFieldSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoicePaymentFieldSqlDao = dbi.onDemand(BusinessInvoicePaymentFieldSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String paymentId = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 30);
+        final String value = UUID.randomUUID().toString();
+
+        // Verify initial state
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId).size(), 0);
+        Assert.assertEquals(invoicePaymentFieldSqlDao.removeField(paymentId, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(invoicePaymentFieldSqlDao.addField(paymentId, name, value), 1);
+        final List<BusinessInvoicePaymentField> fieldsForInvoicePayment = invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId);
+        Assert.assertEquals(fieldsForInvoicePayment.size(), 1);
+
+        // Retrieve it
+        final BusinessInvoicePaymentField invoicePaymentField = fieldsForInvoicePayment.get(0);
+        Assert.assertEquals(invoicePaymentField.getPaymentId().toString(), paymentId);
+        Assert.assertEquals(invoicePaymentField.getName(), name);
+        Assert.assertEquals(invoicePaymentField.getValue(), value);
+
+        // Delete it
+        Assert.assertEquals(invoicePaymentFieldSqlDao.removeField(paymentId, name), 1);
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String paymentId1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 30);
+        final String paymentId2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 30);
+
+        // Add a field to both invoice payments
+        Assert.assertEquals(invoicePaymentFieldSqlDao.addField(paymentId1, name1, UUID.randomUUID().toString()), 1);
+        Assert.assertEquals(invoicePaymentFieldSqlDao.addField(paymentId2, name2, UUID.randomUUID().toString()), 1);
+
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId1).size(), 1);
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId2).size(), 1);
+
+        // Remove the field for the first invoice payment
+        Assert.assertEquals(invoicePaymentFieldSqlDao.removeField(paymentId1, name1), 1);
+
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId1).size(), 0);
+        Assert.assertEquals(invoicePaymentFieldSqlDao.getFieldsForInvoicePayment(paymentId2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoicePaymentFieldSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentSqlDao.java
new file mode 100644
index 0000000..94a8066
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentSqlDao.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoicePayment;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoicePaymentSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoicePaymentSqlDao invoicePaymentSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoicePaymentSqlDao = dbi.onDemand(BusinessInvoicePaymentSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final UUID attemptId = UUID.randomUUID();
+        final String accountKey = UUID.randomUUID().toString();
+        final BusinessInvoicePayment invoicePayment = createInvoicePayment(attemptId, accountKey);
+
+        // Verify initial state
+        Assert.assertNull(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()));
+        Assert.assertEquals(invoicePaymentSqlDao.deleteInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()), 0);
+
+        // Add the invoice payment
+        Assert.assertEquals(invoicePaymentSqlDao.createInvoicePayment(invoicePayment), 1);
+
+        // Retrieve it
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()), invoicePayment);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment.getAccountKey()).get(0), invoicePayment);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment.getPaymentId().toString()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment.getPaymentId().toString()).get(0), invoicePayment);
+
+        // Update and retrieve it
+        invoicePayment.setCardType(UUID.randomUUID().toString().substring(0, 20));
+        Assert.assertEquals(invoicePaymentSqlDao.updateInvoicePaymentForPaymentAttempt(invoicePayment), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()), invoicePayment);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment.getAccountKey()).get(0), invoicePayment);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment.getPaymentId().toString()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment.getPaymentId().toString()).get(0), invoicePayment);
+
+        // Delete it
+        Assert.assertEquals(invoicePaymentSqlDao.deleteInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()), 1);
+        Assert.assertNull(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment.getAttemptId().toString()));
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment.getAccountKey()).size(), 0);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment.getPaymentId().toString()).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final UUID attemptId1 = UUID.randomUUID();
+        final String accountKey1 = UUID.randomUUID().toString();
+        final BusinessInvoicePayment invoicePayment1 = createInvoicePayment(attemptId1, accountKey1);
+        final UUID attemptId2 = UUID.randomUUID();
+        final String accountKey2 = UUID.randomUUID().toString();
+        final BusinessInvoicePayment invoicePayment2 = createInvoicePayment(attemptId2, accountKey2);
+
+        // Create both invoice payments
+        Assert.assertEquals(invoicePaymentSqlDao.createInvoicePayment(invoicePayment1), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.createInvoicePayment(invoicePayment2), 1);
+
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment1.getAttemptId().toString()), invoicePayment1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment2.getAttemptId().toString()), invoicePayment2);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment1.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment2.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment1.getPaymentId().toString()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment2.getPaymentId().toString()).size(), 1);
+
+        // Remove the first invoice payment
+        Assert.assertEquals(invoicePaymentSqlDao.deleteInvoicePaymentForPaymentAttempt(invoicePayment1.getAttemptId().toString()), 1);
+
+        Assert.assertNull(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment1.getAttemptId().toString()));
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentForPaymentAttempt(invoicePayment2.getAttemptId().toString()), invoicePayment2);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment1.getAccountKey()).size(), 0);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForAccount(invoicePayment2.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment1.getPaymentId().toString()).size(), 0);
+        Assert.assertEquals(invoicePaymentSqlDao.getInvoicePaymentsForPayment(invoicePayment2.getPaymentId().toString()).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoicePaymentSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+
+    private BusinessInvoicePayment createInvoicePayment(final UUID attemptId, final String accountKey) {
+        final BigDecimal amount = BigDecimal.ONE;
+        final String cardCountry = UUID.randomUUID().toString().substring(0, 20);
+        final String cardType = UUID.randomUUID().toString().substring(0, 20);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.BRL;
+        final DateTime effectiveDate = new DateTime(DateTimeZone.UTC);
+        final UUID invoiceId = UUID.randomUUID();
+        final String paymentError = UUID.randomUUID().toString();
+        final UUID paymentId = UUID.randomUUID();
+        final String paymentMethod = UUID.randomUUID().toString().substring(0, 20);
+        final String paymentType = UUID.randomUUID().toString().substring(0, 20);
+        final String pluginName = UUID.randomUUID().toString().substring(0, 20);
+        final String processingStatus = UUID.randomUUID().toString();
+        final BigDecimal requestedAmount = BigDecimal.ZERO;
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+
+        return new BusinessInvoicePayment(accountKey, amount, attemptId,
+                                          cardCountry, cardType, createdDate,
+                                          currency, effectiveDate, invoiceId,
+                                          paymentError, paymentId, paymentMethod,
+                                          paymentType, pluginName, processingStatus,
+                                          requestedAmount, updatedDate);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentTagSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentTagSqlDao.java
new file mode 100644
index 0000000..5ae8f48
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoicePaymentTagSqlDao.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoicePaymentTag;
+
+public class TestBusinessInvoicePaymentTagSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoicePaymentTagSqlDao invoicePaymentTagSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoicePaymentTagSqlDao = dbi.onDemand(BusinessInvoicePaymentTagSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String paymentId = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+
+        // Verify initial state
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId).size(), 0);
+        Assert.assertEquals(invoicePaymentTagSqlDao.removeTag(paymentId, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(invoicePaymentTagSqlDao.addTag(paymentId, name), 1);
+        final List<BusinessInvoicePaymentTag> tagsForInvoicePayment = invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId);
+        Assert.assertEquals(tagsForInvoicePayment.size(), 1);
+
+        // Retrieve it
+        final BusinessInvoicePaymentTag invoicePaymentTag = tagsForInvoicePayment.get(0);
+        Assert.assertEquals(invoicePaymentTag.getPaymentId().toString(), paymentId);
+        Assert.assertEquals(invoicePaymentTag.getName(), name);
+
+        // Delete it
+        Assert.assertEquals(invoicePaymentTagSqlDao.removeTag(paymentId, name), 1);
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String paymentId1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 20);
+        final String paymentId2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 20);
+
+        // Add a tag to both invoice payments
+        Assert.assertEquals(invoicePaymentTagSqlDao.addTag(paymentId1, name1), 1);
+        Assert.assertEquals(invoicePaymentTagSqlDao.addTag(paymentId2, name2), 1);
+
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId1).size(), 1);
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId2).size(), 1);
+
+        // Remove the tag for the first invoice payment
+        Assert.assertEquals(invoicePaymentTagSqlDao.removeTag(paymentId1, name1), 1);
+
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId1).size(), 0);
+        Assert.assertEquals(invoicePaymentTagSqlDao.getTagsForInvoicePayment(paymentId2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoicePaymentTagSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceSqlDao.java
new file mode 100644
index 0000000..8373f81
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceSqlDao.java
@@ -0,0 +1,120 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoiceSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoiceSqlDao invoiceSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoiceSqlDao = dbi.onDemand(BusinessInvoiceSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final UUID invoiceId = UUID.randomUUID();
+        final String accountKey = UUID.randomUUID().toString();
+        final BusinessInvoice invoice = createInvoice(invoiceId, accountKey);
+
+        // Verify initial state
+        Assert.assertNull(invoiceSqlDao.getInvoice(invoice.getInvoiceId().toString()));
+        Assert.assertEquals(invoiceSqlDao.deleteInvoice(invoice.getInvoiceId().toString()), 0);
+
+        // Add the invoice
+        Assert.assertEquals(invoiceSqlDao.createInvoice(invoice), 1);
+
+        // Retrieve it
+        Assert.assertEquals(invoiceSqlDao.getInvoice(invoice.getInvoiceId().toString()), invoice);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(invoice.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(invoice.getAccountKey()).get(0), invoice);
+
+        // Update and retrieve it
+        invoice.setInvoiceDate(new DateTime(DateTimeZone.UTC));
+        Assert.assertEquals(invoiceSqlDao.updateInvoice(invoice), 1);
+        Assert.assertEquals(invoiceSqlDao.getInvoice(invoice.getInvoiceId().toString()), invoice);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(invoice.getAccountKey()).size(), 1);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(invoice.getAccountKey()).get(0), invoice);
+
+        // Delete it
+        Assert.assertEquals(invoiceSqlDao.deleteInvoice(invoice.getInvoiceId().toString()), 1);
+        Assert.assertNull(invoiceSqlDao.getInvoice(invoice.getInvoiceId().toString()));
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(invoice.getAccountKey()).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final UUID invoiceId1 = UUID.randomUUID();
+        final String accountKey1 = UUID.randomUUID().toString();
+        final BusinessInvoice invoice1 = createInvoice(invoiceId1, accountKey1);
+        final UUID invoiceId2 = UUID.randomUUID();
+        final String accountKey2 = UUID.randomUUID().toString();
+        final BusinessInvoice invoice2 = createInvoice(invoiceId2, accountKey2);
+
+        // Create both invoices
+        Assert.assertEquals(invoiceSqlDao.createInvoice(invoice1), 1);
+        Assert.assertEquals(invoiceSqlDao.createInvoice(invoice2), 1);
+
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(accountKey1).size(), 1);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(accountKey2).size(), 1);
+
+        // Remove the first invoice
+        Assert.assertEquals(invoiceSqlDao.deleteInvoice(invoice1.getInvoiceId().toString()), 1);
+
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(accountKey1).size(), 0);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(accountKey2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoiceSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+
+    private BusinessInvoice createInvoice(final UUID invoiceId, final String accountKey) {
+        final BigDecimal amountCharged = BigDecimal.ZERO;
+        final BigDecimal amountCredited = BigDecimal.ONE;
+        final BigDecimal amountPaid = BigDecimal.TEN;
+        final BigDecimal balance = BigDecimal.valueOf(123L);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.MXN;
+        final DateTime invoiceDate = new DateTime(DateTimeZone.UTC);
+        final DateTime targetDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+
+        return new BusinessInvoice(accountKey, amountCharged, amountCredited, amountPaid, balance,
+                                   createdDate, currency, invoiceDate, invoiceId, targetDate, updatedDate);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceTagSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceTagSqlDao.java
new file mode 100644
index 0000000..24ee93f
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessInvoiceTagSqlDao.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessInvoiceTag;
+
+public class TestBusinessInvoiceTagSqlDao extends TestWithEmbeddedDB {
+    private BusinessInvoiceTagSqlDao invoiceTagSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        invoiceTagSqlDao = dbi.onDemand(BusinessInvoiceTagSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String invoiceId = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+
+        // Verify initial state
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId).size(), 0);
+        Assert.assertEquals(invoiceTagSqlDao.removeTag(invoiceId, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(invoiceTagSqlDao.addTag(invoiceId, name), 1);
+        final List<BusinessInvoiceTag> tagsForInvoice = invoiceTagSqlDao.getTagsForInvoice(invoiceId);
+        Assert.assertEquals(tagsForInvoice.size(), 1);
+
+        // Retrieve it
+        final BusinessInvoiceTag invoiceTag = tagsForInvoice.get(0);
+        Assert.assertEquals(invoiceTag.getInvoiceId().toString(), invoiceId);
+        Assert.assertEquals(invoiceTag.getName(), name);
+
+        // Delete it
+        Assert.assertEquals(invoiceTagSqlDao.removeTag(invoiceId, name), 1);
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String invoiceId1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 20);
+        final String invoiceId2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 20);
+
+        // Add a tag to both invoices
+        Assert.assertEquals(invoiceTagSqlDao.addTag(invoiceId1, name1), 1);
+        Assert.assertEquals(invoiceTagSqlDao.addTag(invoiceId2, name2), 1);
+
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId1).size(), 1);
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId2).size(), 1);
+
+        // Remove the tag for the first invoice
+        Assert.assertEquals(invoiceTagSqlDao.removeTag(invoiceId1, name1), 1);
+
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId1).size(), 0);
+        Assert.assertEquals(invoiceTagSqlDao.getTagsForInvoice(invoiceId2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            invoiceTagSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessOverdueStatusSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessOverdueStatusSqlDao.java
new file mode 100644
index 0000000..c16ab3c
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessOverdueStatusSqlDao.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessOverdueStatus;
+
+public class TestBusinessOverdueStatusSqlDao extends TestWithEmbeddedDB {
+    private BusinessOverdueStatusSqlDao overdueStatusSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        overdueStatusSqlDao = dbi.onDemand(BusinessOverdueStatusSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCreate() throws Exception {
+        final String externalKey = UUID.randomUUID().toString();
+        final BusinessOverdueStatus firstOverdueStatus = createOverdueStatus(externalKey);
+
+        // Verify initial state
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).size(), 0);
+
+        // Add the overdue status
+        Assert.assertEquals(overdueStatusSqlDao.createOverdueStatus(firstOverdueStatus), 1);
+
+        // Retrieve it
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).size(), 1);
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).get(0), firstOverdueStatus);
+
+        // Add a second one
+        final BusinessOverdueStatus secondOverdueStatus = createOverdueStatus(externalKey);
+        Assert.assertEquals(overdueStatusSqlDao.createOverdueStatus(secondOverdueStatus), 1);
+
+        // Retrieve both
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).size(), 2);
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).get(0), firstOverdueStatus);
+        Assert.assertEquals(overdueStatusSqlDao.getOverdueStatusesForBundle(externalKey).get(1), secondOverdueStatus);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            overdueStatusSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+
+    private BusinessOverdueStatus createOverdueStatus(final String externalKey) {
+        final DateTime endDate = new DateTime(DateTimeZone.UTC);
+        final DateTime startDate = new DateTime(DateTimeZone.UTC);
+        final String status = UUID.randomUUID().toString();
+
+        return new BusinessOverdueStatus(endDate, externalKey, startDate, status);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionFieldSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionFieldSqlDao.java
new file mode 100644
index 0000000..5233229
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionFieldSqlDao.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionField;
+
+public class TestBusinessSubscriptionTransitionFieldSqlDao extends TestWithEmbeddedDB {
+    private BusinessSubscriptionTransitionFieldSqlDao subscriptionTransitionFieldSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        subscriptionTransitionFieldSqlDao = dbi.onDemand(BusinessSubscriptionTransitionFieldSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String externalKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 30);
+        final String value = UUID.randomUUID().toString();
+
+        // Verify initial state
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey).size(), 0);
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.removeField(externalKey, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.addField(externalKey, name, value), 1);
+        final List<BusinessSubscriptionTransitionField> fieldsForBusinessSubscriptionTransition = subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey);
+        Assert.assertEquals(fieldsForBusinessSubscriptionTransition.size(), 1);
+
+        // Retrieve it
+        final BusinessSubscriptionTransitionField subscriptionTransitionField = fieldsForBusinessSubscriptionTransition.get(0);
+        Assert.assertEquals(subscriptionTransitionField.getExternalKey(), externalKey);
+        Assert.assertEquals(subscriptionTransitionField.getName(), name);
+        Assert.assertEquals(subscriptionTransitionField.getValue(), value);
+
+        // Delete it
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.removeField(externalKey, name), 1);
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String externalKey1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 30);
+        final String externalKey2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 30);
+
+        // Add a field to both transitions
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.addField(externalKey1, name1, UUID.randomUUID().toString()), 1);
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.addField(externalKey2, name2, UUID.randomUUID().toString()), 1);
+
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey1).size(), 1);
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey2).size(), 1);
+
+        // Remove the field for the first transition
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.removeField(externalKey1, name1), 1);
+
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey1).size(), 0);
+        Assert.assertEquals(subscriptionTransitionFieldSqlDao.getFieldsForBusinessSubscriptionTransition(externalKey2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            subscriptionTransitionFieldSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionTagSqlDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionTagSqlDao.java
new file mode 100644
index 0000000..8b4740f
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestBusinessSubscriptionTransitionTagSqlDao.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionTag;
+
+public class TestBusinessSubscriptionTransitionTagSqlDao extends TestWithEmbeddedDB {
+    private BusinessSubscriptionTransitionTagSqlDao subscriptionTransitionTagSqlDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        subscriptionTransitionTagSqlDao = dbi.onDemand(BusinessSubscriptionTransitionTagSqlDao.class);
+    }
+
+    @Test(groups = "slow")
+    public void testCRUD() throws Exception {
+        final String externalKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+
+        // Verify initial state
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey).size(), 0);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.removeTag(externalKey, name), 0);
+
+        // Add an entry
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.addTag(externalKey, name), 1);
+        final List<BusinessSubscriptionTransitionTag> tagsForBusinessSubscriptionTransition = subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey);
+        Assert.assertEquals(tagsForBusinessSubscriptionTransition.size(), 1);
+
+        // Retrieve it
+        final BusinessSubscriptionTransitionTag subscriptionTransitionTag = tagsForBusinessSubscriptionTransition.get(0);
+        Assert.assertEquals(subscriptionTransitionTag.getExternalKey(), externalKey);
+        Assert.assertEquals(subscriptionTransitionTag.getName(), name);
+
+        // Delete it
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.removeTag(externalKey, name), 1);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSegmentation() throws Exception {
+        final String externalKey1 = UUID.randomUUID().toString();
+        final String name1 = UUID.randomUUID().toString().substring(0, 20);
+        final String externalKey2 = UUID.randomUUID().toString();
+        final String name2 = UUID.randomUUID().toString().substring(0, 20);
+
+        // Add a tag to both transitions
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.addTag(externalKey1, name1), 1);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.addTag(externalKey2, name2), 1);
+
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey1).size(), 1);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey2).size(), 1);
+
+        // Remove the tag for the first transition
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.removeTag(externalKey1, name1), 1);
+
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey1).size(), 0);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey2).size(), 1);
+    }
+
+    @Test(groups = "slow")
+    public void testHealthCheck() throws Exception {
+        // HealthCheck test to make sure MySQL is setup properly
+        try {
+            subscriptionTransitionTagSqlDao.test();
+        } catch (Throwable t) {
+            Assert.fail(t.toString());
+        }
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestDefaultAnalyticsDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestDefaultAnalyticsDao.java
new file mode 100644
index 0000000..7ab6a66
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestDefaultAnalyticsDao.java
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.dao;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.TestWithEmbeddedDB;
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestDefaultAnalyticsDao extends TestWithEmbeddedDB {
+    private BusinessAccountSqlDao accountSqlDao;
+    private BusinessInvoiceSqlDao invoiceSqlDao;
+    private BusinessInvoiceItemSqlDao invoiceItemSqlDao;
+    private AnalyticsDao analyticsDao;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        accountSqlDao = dbi.onDemand(BusinessAccountSqlDao.class);
+        final BusinessSubscriptionTransitionSqlDao subscriptionTransitionSqlDao = dbi.onDemand(BusinessSubscriptionTransitionSqlDao.class);
+        invoiceSqlDao = dbi.onDemand(BusinessInvoiceSqlDao.class);
+        invoiceItemSqlDao = dbi.onDemand(BusinessInvoiceItemSqlDao.class);
+        final BusinessAccountTagSqlDao accountTagSqlDao = dbi.onDemand(BusinessAccountTagSqlDao.class);
+        analyticsDao = new DefaultAnalyticsDao(accountSqlDao, subscriptionTransitionSqlDao, invoiceSqlDao, invoiceItemSqlDao, accountTagSqlDao);
+    }
+
+    @Test(groups = "slow")
+    public void testCreateInvoice() throws Exception {
+        // Create and verify the initial state
+        BusinessAccount account = new BusinessAccount(UUID.randomUUID().toString(), UUID.randomUUID().toString(),
+                                                      BigDecimal.ONE, new DateTime(DateTimeZone.UTC), BigDecimal.TEN,
+                                                      "ERROR_NOT_ENOUGH_FUNDS", "CreditCard", "Visa", "FRANCE");
+        Assert.assertEquals(accountSqlDao.createAccount(account), 1);
+        Assert.assertEquals(invoiceSqlDao.getInvoicesForAccount(account.getKey()).size(), 0);
+        account = accountSqlDao.getAccount(account.getKey());
+
+        // Generate the invoices
+        final BusinessInvoice invoice = createInvoice(account.getKey());
+        final List<BusinessInvoiceItem> invoiceItems = new ArrayList<BusinessInvoiceItem>();
+        for (int i = 0; i < 10; i++) {
+            invoiceItems.add(createInvoiceItem(invoice.getInvoiceId(), BigDecimal.valueOf(1242 + i)));
+        }
+        analyticsDao.createInvoice(account.getKey(), invoice, invoiceItems);
+
+        // Verify the final state
+        final List<BusinessInvoice> invoicesForAccount = invoiceSqlDao.getInvoicesForAccount(account.getKey());
+        Assert.assertEquals(invoicesForAccount.size(), 1);
+        Assert.assertEquals(invoicesForAccount.get(0).getInvoiceId(), invoice.getInvoiceId());
+
+        Assert.assertEquals(invoiceItemSqlDao.getInvoiceItemsForInvoice(invoice.getInvoiceId().toString()).size(), 10);
+
+        final BusinessAccount finalAccount = accountSqlDao.getAccount(account.getKey());
+        Assert.assertEquals(finalAccount.getCreatedDt(), account.getCreatedDt());
+        Assert.assertTrue(finalAccount.getUpdatedDt().isAfter(account.getCreatedDt()));
+        Assert.assertTrue(finalAccount.getUpdatedDt().isAfter(account.getUpdatedDt()));
+        Assert.assertTrue(finalAccount.getLastInvoiceDate().equals(invoice.getInvoiceDate()));
+        // invoice.getBalance() is not the sum of all the items here - but in practice it will be
+        Assert.assertEquals(finalAccount.getTotalInvoiceBalance(), account.getTotalInvoiceBalance().add(invoice.getBalance()));
+    }
+
+    private BusinessInvoice createInvoice(final String accountKey) {
+        final BigDecimal amountCharged = BigDecimal.ZERO;
+        final BigDecimal amountCredited = BigDecimal.ONE;
+        final BigDecimal amountPaid = BigDecimal.TEN;
+        final BigDecimal balance = BigDecimal.valueOf(123L);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.MXN;
+        final DateTime invoiceDate = new DateTime(DateTimeZone.UTC);
+        final UUID invoiceId = UUID.randomUUID();
+        final DateTime targetDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+
+        return new BusinessInvoice(accountKey, amountCharged, amountCredited, amountPaid, balance,
+                                   createdDate, currency, invoiceDate, invoiceId, targetDate, updatedDate);
+    }
+
+    private BusinessInvoiceItem createInvoiceItem(final UUID invoiceId, final BigDecimal amount) {
+        final String billingPeriod = UUID.randomUUID().toString().substring(0, 20);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.AUD;
+        final DateTime endDate = new DateTime(DateTimeZone.UTC);
+        final String externalKey = UUID.randomUUID().toString();
+        final UUID itemId = UUID.randomUUID();
+        final String itemType = UUID.randomUUID().toString().substring(0, 20);
+        final String phase = UUID.randomUUID().toString().substring(0, 20);
+        final String productCategory = UUID.randomUUID().toString().substring(0, 20);
+        final String productName = UUID.randomUUID().toString().substring(0, 20);
+        final String productType = UUID.randomUUID().toString().substring(0, 20);
+        final String slug = UUID.randomUUID().toString().substring(0, 20);
+        final DateTime startDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+
+        return new BusinessInvoiceItem(amount, billingPeriod, createdDate, currency,
+                                       endDate, externalKey, invoiceId, itemId, itemType,
+                                       phase, productCategory, productName, productType,
+                                       slug, startDate, updatedDate);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountField.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountField.java
new file mode 100644
index 0000000..ebe3c9e
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountField.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessAccountField extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString();
+        final String value = UUID.randomUUID().toString();
+        final BusinessAccountField accountField = new BusinessAccountField(accountKey,
+                                                                           name,
+                                                                           value);
+        Assert.assertSame(accountField, accountField);
+        Assert.assertEquals(accountField, accountField);
+        Assert.assertTrue(accountField.equals(accountField));
+        Assert.assertEquals(accountField.getAccountKey(), accountKey);
+        Assert.assertEquals(accountField.getName(), name);
+        Assert.assertEquals(accountField.getValue(), value);
+
+        final BusinessAccountField otherAccountField = new BusinessAccountField(UUID.randomUUID().toString(),
+                                                                                UUID.randomUUID().toString(),
+                                                                                UUID.randomUUID().toString());
+        Assert.assertFalse(accountField.equals(otherAccountField));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountTag.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountTag.java
new file mode 100644
index 0000000..1d9c072
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessAccountTag.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessAccountTag extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString();
+        final BusinessAccountTag accountTag = new BusinessAccountTag(accountKey, name);
+        Assert.assertSame(accountTag, accountTag);
+        Assert.assertEquals(accountTag, accountTag);
+        Assert.assertTrue(accountTag.equals(accountTag));
+        Assert.assertEquals(accountTag.getAccountKey(), accountKey);
+        Assert.assertEquals(accountTag.getName(), name);
+
+        final BusinessAccountTag otherAccountTag = new BusinessAccountTag(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+        Assert.assertFalse(accountTag.equals(otherAccountTag));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoice.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoice.java
new file mode 100644
index 0000000..869284d
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoice.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoice extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final BigDecimal amountCharged = BigDecimal.ZERO;
+        final BigDecimal amountCredited = BigDecimal.ONE;
+        final BigDecimal amountPaid = BigDecimal.TEN;
+        final BigDecimal balance = BigDecimal.valueOf(123L);
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.MXN;
+        final DateTime invoiceDate = new DateTime(DateTimeZone.UTC);
+        final UUID invoiceId = UUID.randomUUID();
+        final DateTime targetDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+        final BusinessInvoice invoice = new BusinessInvoice(accountKey, amountCharged, amountCredited, amountPaid, balance,
+                                                            createdDate, currency, invoiceDate, invoiceId, targetDate, updatedDate);
+        Assert.assertSame(invoice, invoice);
+        Assert.assertEquals(invoice, invoice);
+        Assert.assertTrue(invoice.equals(invoice));
+        Assert.assertEquals(invoice.getAccountKey(), accountKey);
+        Assert.assertEquals(invoice.getAmountCharged(), amountCharged);
+        Assert.assertEquals(invoice.getAmountCredited(), amountCredited);
+        Assert.assertEquals(invoice.getAmountPaid(), amountPaid);
+        Assert.assertEquals(invoice.getBalance(), balance);
+        Assert.assertEquals(invoice.getCreatedDate(), createdDate);
+        Assert.assertEquals(invoice.getCurrency(), currency);
+        Assert.assertEquals(invoice.getInvoiceDate(), invoiceDate);
+        Assert.assertEquals(invoice.getInvoiceId(), invoiceId);
+        Assert.assertEquals(invoice.getTargetDate(), targetDate);
+        Assert.assertEquals(invoice.getUpdatedDate(), updatedDate);
+
+        final BusinessInvoice otherInvoice = new BusinessInvoice(null, null, null, null, null, createdDate, null,
+                                                                 null, invoiceId, null, null);
+        Assert.assertFalse(invoice.equals(otherInvoice));
+
+        // Test setters
+        otherInvoice.setAccountKey(accountKey);
+        otherInvoice.setAmountCharged(amountCharged);
+        otherInvoice.setAmountCredited(amountCredited);
+        otherInvoice.setAmountPaid(amountPaid);
+        otherInvoice.setBalance(balance);
+        otherInvoice.setCurrency(currency);
+        otherInvoice.setInvoiceDate(invoiceDate);
+        otherInvoice.setTargetDate(targetDate);
+        otherInvoice.setUpdatedDate(updatedDate);
+        Assert.assertTrue(invoice.equals(otherInvoice));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceField.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceField.java
new file mode 100644
index 0000000..b2c0937
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceField.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessInvoiceField extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final UUID invoiceId = UUID.randomUUID();
+        final String name = UUID.randomUUID().toString();
+        final String value = UUID.randomUUID().toString();
+        final BusinessInvoiceField invoiceField = new BusinessInvoiceField(invoiceId,
+                                                                           name,
+                                                                           value);
+        Assert.assertSame(invoiceField, invoiceField);
+        Assert.assertEquals(invoiceField, invoiceField);
+        Assert.assertTrue(invoiceField.equals(invoiceField));
+        Assert.assertEquals(invoiceField.getInvoiceId(), invoiceId);
+        Assert.assertEquals(invoiceField.getName(), name);
+        Assert.assertEquals(invoiceField.getValue(), value);
+
+        final BusinessInvoiceField otherInvoiceField = new BusinessInvoiceField(UUID.randomUUID(),
+                                                                                UUID.randomUUID().toString(),
+                                                                                UUID.randomUUID().toString());
+        Assert.assertFalse(invoiceField.equals(otherInvoiceField));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceItem.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceItem.java
new file mode 100644
index 0000000..e4460f6
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceItem.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoiceItem extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final BigDecimal amount = BigDecimal.TEN;
+        final String billingPeriod = UUID.randomUUID().toString();
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.AUD;
+        final DateTime endDate = new DateTime(DateTimeZone.UTC);
+        final String externalKey = UUID.randomUUID().toString();
+        final UUID invoiceId = UUID.randomUUID();
+        final UUID itemId = UUID.randomUUID();
+        final String itemType = UUID.randomUUID().toString();
+        final String phase = UUID.randomUUID().toString();
+        final String productCategory = UUID.randomUUID().toString();
+        final String productName = UUID.randomUUID().toString();
+        final String productType = UUID.randomUUID().toString();
+        final String slug = UUID.randomUUID().toString();
+        final DateTime startDate = new DateTime(DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+        final BusinessInvoiceItem invoiceItem = new BusinessInvoiceItem(amount, billingPeriod, createdDate, currency,
+                                                                        endDate, externalKey, invoiceId, itemId, itemType,
+                                                                        phase, productCategory, productName, productType,
+                                                                        slug, startDate, updatedDate);
+        Assert.assertSame(invoiceItem, invoiceItem);
+        Assert.assertEquals(invoiceItem, invoiceItem);
+        Assert.assertTrue(invoiceItem.equals(invoiceItem));
+        Assert.assertEquals(invoiceItem.getAmount(), amount);
+        Assert.assertEquals(invoiceItem.getBillingPeriod(), billingPeriod);
+        Assert.assertEquals(invoiceItem.getCreatedDate(), createdDate);
+        Assert.assertEquals(invoiceItem.getCurrency(), currency);
+        Assert.assertEquals(invoiceItem.getEndDate(), endDate);
+        Assert.assertEquals(invoiceItem.getExternalKey(), externalKey);
+        Assert.assertEquals(invoiceItem.getInvoiceId(), invoiceId);
+        Assert.assertEquals(invoiceItem.getItemId(), itemId);
+        Assert.assertEquals(invoiceItem.getItemType(), itemType);
+        Assert.assertEquals(invoiceItem.getPhase(), phase);
+        Assert.assertEquals(invoiceItem.getProductCategory(), productCategory);
+        Assert.assertEquals(invoiceItem.getProductName(), productName);
+        Assert.assertEquals(invoiceItem.getProductType(), productType);
+        Assert.assertEquals(invoiceItem.getSlug(), slug);
+        Assert.assertEquals(invoiceItem.getStartDate(), startDate);
+        Assert.assertEquals(invoiceItem.getUpdatedDate(), updatedDate);
+
+        final BusinessInvoiceItem otherInvoiceItem = new BusinessInvoiceItem(null, null, createdDate, null, null, null, null, itemId,
+                                                                             null, null, null, null, null, null, null, null);
+        Assert.assertFalse(invoiceItem.equals(otherInvoiceItem));
+
+        // Test setters
+        otherInvoiceItem.setAmount(amount);
+        otherInvoiceItem.setBillingPeriod(billingPeriod);
+        otherInvoiceItem.setCurrency(currency);
+        otherInvoiceItem.setEndDate(endDate);
+        otherInvoiceItem.setExternalKey(externalKey);
+        otherInvoiceItem.setInvoiceId(invoiceId);
+        otherInvoiceItem.setItemType(itemType);
+        otherInvoiceItem.setPhase(phase);
+        otherInvoiceItem.setProductCategory(productCategory);
+        otherInvoiceItem.setProductName(productName);
+        otherInvoiceItem.setProductType(productType);
+        otherInvoiceItem.setSlug(slug);
+        otherInvoiceItem.setStartDate(startDate);
+        otherInvoiceItem.setUpdatedDate(updatedDate);
+        Assert.assertTrue(invoiceItem.equals(otherInvoiceItem));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePayment.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePayment.java
new file mode 100644
index 0000000..709db0f
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePayment.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.math.BigDecimal;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+import com.ning.billing.catalog.api.Currency;
+
+public class TestBusinessInvoicePayment extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String accountKey = UUID.randomUUID().toString();
+        final BigDecimal amount = BigDecimal.ONE;
+        final UUID attemptId = UUID.randomUUID();
+        final String cardCountry = UUID.randomUUID().toString();
+        final String cardType = UUID.randomUUID().toString();
+        final DateTime createdDate = new DateTime(DateTimeZone.UTC);
+        final Currency currency = Currency.BRL;
+        final DateTime effectiveDate = new DateTime(DateTimeZone.UTC);
+        final UUID invoiceId = UUID.randomUUID();
+        final String paymentError = UUID.randomUUID().toString();
+        final UUID paymentId = UUID.randomUUID();
+        final String paymentMethod = UUID.randomUUID().toString();
+        final String paymentType = UUID.randomUUID().toString();
+        final String pluginName = UUID.randomUUID().toString();
+        final String processingStatus = UUID.randomUUID().toString();
+        final BigDecimal requestedAmount = BigDecimal.ZERO;
+        final DateTime updatedDate = new DateTime(DateTimeZone.UTC);
+        final BusinessInvoicePayment invoicePayment = new BusinessInvoicePayment(accountKey, amount, attemptId,
+                                                                                 cardCountry, cardType, createdDate,
+                                                                                 currency, effectiveDate, invoiceId,
+                                                                                 paymentError, paymentId, paymentMethod,
+                                                                                 paymentType, pluginName, processingStatus,
+                                                                                 requestedAmount, updatedDate);
+        Assert.assertSame(invoicePayment, invoicePayment);
+        Assert.assertEquals(invoicePayment, invoicePayment);
+        Assert.assertTrue(invoicePayment.equals(invoicePayment));
+        Assert.assertEquals(invoicePayment.getAccountKey(), accountKey);
+        Assert.assertEquals(invoicePayment.getAmount(), amount);
+        Assert.assertEquals(invoicePayment.getAttemptId(), attemptId);
+        Assert.assertEquals(invoicePayment.getCardCountry(), cardCountry);
+        Assert.assertEquals(invoicePayment.getCardType(), cardType);
+        Assert.assertEquals(invoicePayment.getCreatedDate(), createdDate);
+        Assert.assertEquals(invoicePayment.getCurrency(), currency);
+        Assert.assertEquals(invoicePayment.getEffectiveDate(), effectiveDate);
+        Assert.assertEquals(invoicePayment.getInvoiceId(), invoiceId);
+        Assert.assertEquals(invoicePayment.getPaymentError(), paymentError);
+        Assert.assertEquals(invoicePayment.getPaymentId(), paymentId);
+        Assert.assertEquals(invoicePayment.getPaymentMethod(), paymentMethod);
+        Assert.assertEquals(invoicePayment.getPaymentType(), paymentType);
+        Assert.assertEquals(invoicePayment.getPluginName(), pluginName);
+        Assert.assertEquals(invoicePayment.getProcessingStatus(), processingStatus);
+        Assert.assertEquals(invoicePayment.getRequestedAmount(), requestedAmount);
+        Assert.assertEquals(invoicePayment.getUpdatedDate(), updatedDate);
+
+        final BusinessInvoicePayment otherInvoicePayment = new BusinessInvoicePayment(null, null, attemptId, null, null, createdDate,
+                                                                                      null, null, null, null, paymentId, null,
+                                                                                      null, null, null, null, null);
+        Assert.assertFalse(invoicePayment.equals(otherInvoicePayment));
+
+        // Test setters
+        otherInvoicePayment.setAccountKey(accountKey);
+        otherInvoicePayment.setAmount(amount);
+        otherInvoicePayment.setCardCountry(cardCountry);
+        otherInvoicePayment.setCardType(cardType);
+        otherInvoicePayment.setCurrency(currency);
+        otherInvoicePayment.setEffectiveDate(effectiveDate);
+        otherInvoicePayment.setInvoiceId(invoiceId);
+        otherInvoicePayment.setPaymentError(paymentError);
+        otherInvoicePayment.setPaymentMethod(paymentMethod);
+        otherInvoicePayment.setPaymentType(paymentType);
+        otherInvoicePayment.setPluginName(pluginName);
+        otherInvoicePayment.setProcessingStatus(processingStatus);
+        otherInvoicePayment.setRequestedAmount(requestedAmount);
+        otherInvoicePayment.setUpdatedDate(updatedDate);
+        Assert.assertTrue(invoicePayment.equals(otherInvoicePayment));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentField.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentField.java
new file mode 100644
index 0000000..2d73c48
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentField.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessInvoicePaymentField extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final UUID paymentId = UUID.randomUUID();
+        final String name = UUID.randomUUID().toString();
+        final String value = UUID.randomUUID().toString();
+        final BusinessInvoicePaymentField invoiceField = new BusinessInvoicePaymentField(paymentId,
+                                                                                         name,
+                                                                                         value);
+        Assert.assertSame(invoiceField, invoiceField);
+        Assert.assertEquals(invoiceField, invoiceField);
+        Assert.assertTrue(invoiceField.equals(invoiceField));
+        Assert.assertEquals(invoiceField.getPaymentId(), paymentId);
+        Assert.assertEquals(invoiceField.getName(), name);
+        Assert.assertEquals(invoiceField.getValue(), value);
+
+        final BusinessInvoicePaymentField otherInvoicePaymentField = new BusinessInvoicePaymentField(UUID.randomUUID(),
+                                                                                                     UUID.randomUUID().toString(),
+                                                                                                     UUID.randomUUID().toString());
+        Assert.assertFalse(invoiceField.equals(otherInvoicePaymentField));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentTag.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentTag.java
new file mode 100644
index 0000000..b5ae515
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoicePaymentTag.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessInvoicePaymentTag extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final UUID paymentId = UUID.randomUUID();
+        final String name = UUID.randomUUID().toString();
+        final BusinessInvoicePaymentTag invoicePaymentTag = new BusinessInvoicePaymentTag(paymentId, name);
+        Assert.assertSame(invoicePaymentTag, invoicePaymentTag);
+        Assert.assertEquals(invoicePaymentTag, invoicePaymentTag);
+        Assert.assertTrue(invoicePaymentTag.equals(invoicePaymentTag));
+        Assert.assertEquals(invoicePaymentTag.getPaymentId(), paymentId);
+        Assert.assertEquals(invoicePaymentTag.getName(), name);
+
+        final BusinessInvoicePaymentTag otherInvoicePaymentTag = new BusinessInvoicePaymentTag(UUID.randomUUID(),
+                                                                                               UUID.randomUUID().toString());
+        Assert.assertFalse(invoicePaymentTag.equals(otherInvoicePaymentTag));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceTag.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceTag.java
new file mode 100644
index 0000000..3e92eae
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessInvoiceTag.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessInvoiceTag extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final UUID invoiceId = UUID.randomUUID();
+        final String name = UUID.randomUUID().toString();
+        final BusinessInvoiceTag invoiceTag = new BusinessInvoiceTag(invoiceId, name);
+        Assert.assertSame(invoiceTag, invoiceTag);
+        Assert.assertEquals(invoiceTag, invoiceTag);
+        Assert.assertTrue(invoiceTag.equals(invoiceTag));
+        Assert.assertEquals(invoiceTag.getInvoiceId(), invoiceId);
+        Assert.assertEquals(invoiceTag.getName(), name);
+
+        final BusinessInvoiceTag otherInvoiceTag = new BusinessInvoiceTag(UUID.randomUUID(), UUID.randomUUID().toString());
+        Assert.assertFalse(invoiceTag.equals(otherInvoiceTag));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessOverdueStatus.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessOverdueStatus.java
new file mode 100644
index 0000000..b92ee12
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessOverdueStatus.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessOverdueStatus extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final DateTime endDate = new DateTime(DateTimeZone.UTC);
+        final String externalKey = UUID.randomUUID().toString();
+        final DateTime startDate = new DateTime(DateTimeZone.UTC);
+        final String status = UUID.randomUUID().toString();
+        final BusinessOverdueStatus overdueStatus = new BusinessOverdueStatus(endDate, externalKey, startDate, status);
+        Assert.assertSame(overdueStatus, overdueStatus);
+        Assert.assertEquals(overdueStatus, overdueStatus);
+        Assert.assertTrue(overdueStatus.equals(overdueStatus));
+        Assert.assertEquals(overdueStatus.getEndDate(), endDate);
+        Assert.assertEquals(overdueStatus.getExternalKey(), externalKey);
+        Assert.assertEquals(overdueStatus.getStartDate(), startDate);
+        Assert.assertEquals(overdueStatus.getStatus(), status);
+
+        final BusinessOverdueStatus otherOverdueStatus = new BusinessOverdueStatus(new DateTime(DateTimeZone.UTC), UUID.randomUUID().toString(),
+                                                                                   new DateTime(DateTimeZone.UTC), UUID.randomUUID().toString());
+        Assert.assertFalse(overdueStatus.equals(otherOverdueStatus));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionField.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionField.java
new file mode 100644
index 0000000..5a1e31b
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionField.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessSubscriptionTransitionField extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String externalKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString();
+        final String value = UUID.randomUUID().toString();
+        final BusinessSubscriptionTransitionField subscriptionTransitionField = new BusinessSubscriptionTransitionField(externalKey,
+                                                                                                                        name,
+                                                                                                                        value);
+        Assert.assertSame(subscriptionTransitionField, subscriptionTransitionField);
+        Assert.assertEquals(subscriptionTransitionField, subscriptionTransitionField);
+        Assert.assertTrue(subscriptionTransitionField.equals(subscriptionTransitionField));
+        Assert.assertEquals(subscriptionTransitionField.getExternalKey(), externalKey);
+        Assert.assertEquals(subscriptionTransitionField.getName(), name);
+        Assert.assertEquals(subscriptionTransitionField.getValue(), value);
+
+        final BusinessSubscriptionTransitionField otherSubscriptionField = new BusinessSubscriptionTransitionField(UUID.randomUUID().toString(),
+                                                                                                                   UUID.randomUUID().toString(),
+                                                                                                                   UUID.randomUUID().toString());
+        Assert.assertFalse(subscriptionTransitionField.equals(otherSubscriptionField));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionTag.java b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionTag.java
new file mode 100644
index 0000000..b5a40d1
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/model/TestBusinessSubscriptionTransitionTag.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics.model;
+
+import java.util.UUID;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.ning.billing.analytics.AnalyticsTestSuite;
+
+public class TestBusinessSubscriptionTransitionTag extends AnalyticsTestSuite {
+    @Test(groups = "fast")
+    public void testEquals() throws Exception {
+        final String externalKey = UUID.randomUUID().toString();
+        final String name = UUID.randomUUID().toString();
+        final BusinessSubscriptionTransitionTag subscriptionTransitionTag = new BusinessSubscriptionTransitionTag(externalKey, name);
+        Assert.assertSame(subscriptionTransitionTag, subscriptionTransitionTag);
+        Assert.assertEquals(subscriptionTransitionTag, subscriptionTransitionTag);
+        Assert.assertTrue(subscriptionTransitionTag.equals(subscriptionTransitionTag));
+        Assert.assertEquals(subscriptionTransitionTag.getExternalKey(), externalKey);
+        Assert.assertEquals(subscriptionTransitionTag.getName(), name);
+
+        final BusinessSubscriptionTransitionTag otherTransitionTag = new BusinessSubscriptionTransitionTag(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+        Assert.assertFalse(subscriptionTransitionTag.equals(otherTransitionTag));
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java b/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java
index 2d29f08..a5cfc42 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java
@@ -74,7 +74,7 @@ public class TestAnalyticsListener extends AnalyticsTestSuite {
     @BeforeMethod(groups = "fast")
     public void setUp() throws Exception {
         final BusinessSubscriptionTransitionRecorder recorder = new BusinessSubscriptionTransitionRecorder(dao, catalogService, new MockEntitlementUserApi(bundleUUID, EXTERNAL_KEY), new MockAccountUserApi(ACCOUNT_KEY, CURRENCY));
-        listener = new AnalyticsListener(recorder, null);
+        listener = new AnalyticsListener(recorder, null, null, null);
     }
 
     @Test(groups = "fast")

diff --git a/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java
new file mode 100644
index 0000000..2eb018b
--- /dev/null
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.analytics;
+
+import java.util.UUID;
+
+import org.mockito.Mockito;
+import org.skife.jdbi.v2.IDBI;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.ning.billing.account.api.Account;
+import com.ning.billing.account.api.AccountUserApi;
+import com.ning.billing.account.api.user.DefaultAccountUserApi;
+import com.ning.billing.account.dao.AccountDao;
+import com.ning.billing.account.dao.AccountEmailDao;
+import com.ning.billing.account.dao.AuditedAccountDao;
+import com.ning.billing.account.dao.AuditedAccountEmailDao;
+import com.ning.billing.analytics.dao.BusinessAccountTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoicePaymentTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessInvoiceTagSqlDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionTagSqlDao;
+import com.ning.billing.catalog.DefaultCatalogService;
+import com.ning.billing.catalog.api.CatalogService;
+import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.catalog.io.VersionedCatalogLoader;
+import com.ning.billing.config.CatalogConfig;
+import com.ning.billing.entitlement.alignment.PlanAligner;
+import com.ning.billing.entitlement.api.user.DefaultEntitlementUserApi;
+import com.ning.billing.entitlement.api.user.DefaultSubscriptionApiService;
+import com.ning.billing.entitlement.api.user.DefaultSubscriptionFactory;
+import com.ning.billing.entitlement.api.user.EntitlementUserApi;
+import com.ning.billing.entitlement.api.user.SubscriptionBundle;
+import com.ning.billing.entitlement.engine.addon.AddonUtils;
+import com.ning.billing.entitlement.engine.dao.AuditedEntitlementDao;
+import com.ning.billing.entitlement.engine.dao.EntitlementDao;
+import com.ning.billing.util.bus.InMemoryBus;
+import com.ning.billing.util.callcontext.CallContext;
+import com.ning.billing.util.callcontext.CallOrigin;
+import com.ning.billing.util.callcontext.DefaultCallContextFactory;
+import com.ning.billing.util.callcontext.UserType;
+import com.ning.billing.util.clock.DefaultClock;
+import com.ning.billing.util.dao.ObjectType;
+import com.ning.billing.util.notificationq.DefaultNotificationQueueService;
+
+public class TestBusinessTagRecorder extends TestWithEmbeddedDB {
+    private BusinessAccountTagSqlDao accountTagSqlDao;
+    private BusinessSubscriptionTransitionTagSqlDao subscriptionTransitionTagSqlDao;
+    private InMemoryBus eventBus;
+    private DefaultCallContextFactory callContextFactory;
+    private AccountUserApi accountUserApi;
+    private EntitlementUserApi entitlementUserApi;
+    private BusinessTagRecorder tagRecorder;
+
+    @BeforeMethod(groups = "slow")
+    public void setUp() throws Exception {
+        final IDBI dbi = helper.getDBI();
+        accountTagSqlDao = dbi.onDemand(BusinessAccountTagSqlDao.class);
+        final BusinessInvoiceTagSqlDao invoiceTagSqlDao = dbi.onDemand(BusinessInvoiceTagSqlDao.class);
+        final BusinessInvoicePaymentTagSqlDao invoicePaymentTagSqlDao = dbi.onDemand(BusinessInvoicePaymentTagSqlDao.class);
+        subscriptionTransitionTagSqlDao = dbi.onDemand(BusinessSubscriptionTransitionTagSqlDao.class);
+        eventBus = new InMemoryBus();
+        final AccountDao accountDao = new AuditedAccountDao(dbi, eventBus);
+        final AccountEmailDao accountEmailDao = new AuditedAccountEmailDao(dbi);
+        final DefaultClock clock = new DefaultClock();
+        callContextFactory = new DefaultCallContextFactory(clock);
+        accountUserApi = new DefaultAccountUserApi(callContextFactory, accountDao, accountEmailDao);
+        final CatalogService catalogService = new DefaultCatalogService(Mockito.mock(CatalogConfig.class), Mockito.mock(VersionedCatalogLoader.class));
+        final AddonUtils addonUtils = new AddonUtils(catalogService);
+        final DefaultNotificationQueueService notificationQueueService = new DefaultNotificationQueueService(dbi, clock);
+        final EntitlementDao entitlementDao = new AuditedEntitlementDao(dbi, clock, addonUtils, notificationQueueService, eventBus);
+        final PlanAligner planAligner = new PlanAligner(catalogService);
+        final DefaultSubscriptionApiService apiService = new DefaultSubscriptionApiService(clock, entitlementDao, catalogService, planAligner);
+        final DefaultSubscriptionFactory subscriptionFactory = new DefaultSubscriptionFactory(apiService, clock, catalogService);
+        entitlementUserApi = new DefaultEntitlementUserApi(clock, entitlementDao, catalogService,
+                                                           apiService, subscriptionFactory, addonUtils);
+        tagRecorder = new BusinessTagRecorder(accountTagSqlDao, invoicePaymentTagSqlDao, invoiceTagSqlDao, subscriptionTransitionTagSqlDao,
+                                              accountUserApi, entitlementUserApi);
+
+        eventBus.start();
+    }
+
+    @AfterMethod(groups = "slow")
+    public void tearDown() throws Exception {
+        eventBus.stop();
+    }
+
+    @Test(groups = "slow")
+    public void testAddAndRemoveTagsForAccount() throws Exception {
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+        final CallContext callContext = callContextFactory.createCallContext(UUID.randomUUID().toString(), CallOrigin.TEST, UserType.TEST);
+        final String accountKey = UUID.randomUUID().toString();
+
+        final Account account = accountUserApi.createAccount(new MockAccount(UUID.randomUUID(), accountKey, Currency.MXN), callContext);
+        final UUID accountId = account.getId();
+
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey).size(), 0);
+        tagRecorder.tagAdded(ObjectType.ACCOUNT, accountId, name);
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey).size(), 1);
+        tagRecorder.tagRemoved(ObjectType.ACCOUNT, accountId, name);
+        Assert.assertEquals(accountTagSqlDao.getTagsForAccount(accountKey).size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testAddAndRemoveTagsForBundle() throws Exception {
+        final String name = UUID.randomUUID().toString().substring(0, 20);
+        final CallContext callContext = callContextFactory.createCallContext(UUID.randomUUID().toString(), CallOrigin.TEST, UserType.TEST);
+        final String externalKey = UUID.randomUUID().toString();
+
+        final Account account = accountUserApi.createAccount(new MockAccount(UUID.randomUUID(), UUID.randomUUID().toString(), Currency.MXN), callContext);
+        final SubscriptionBundle bundle = entitlementUserApi.createBundleForAccount(account.getId(), externalKey, callContext);
+        final UUID bundleId = bundle.getId();
+
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey).size(), 0);
+        tagRecorder.tagAdded(ObjectType.BUNDLE, bundleId, name);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey).size(), 1);
+        tagRecorder.tagRemoved(ObjectType.BUNDLE, bundleId, name);
+        Assert.assertEquals(subscriptionTransitionTagSqlDao.getTagsForBusinessSubscriptionTransition(externalKey).size(), 0);
+    }
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/TestWithEmbeddedDB.java b/analytics/src/test/java/com/ning/billing/analytics/TestWithEmbeddedDB.java
index 8feb144..9f60771 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/TestWithEmbeddedDB.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestWithEmbeddedDB.java
@@ -47,7 +47,7 @@ public abstract class TestWithEmbeddedDB extends AnalyticsTestSuite {
         helper.cleanupAllTables();
     }
 
-    @BeforeMethod(alwaysRun = true)
+    @BeforeMethod(groups = "slow")
     public void cleanup() {
         try {
             helper.cleanupAllTables();

diff --git a/beatrix/pom.xml b/beatrix/pom.xml
index 96d50b9..78690e8 100644
--- a/beatrix/pom.xml
+++ b/beatrix/pom.xml
@@ -108,6 +108,11 @@
         </dependency>
         <dependency>
             <groupId>com.ning.billing</groupId>
+            <artifactId>killbill-analytics</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.ning.billing</groupId>
             <artifactId>killbill-overdue</artifactId>
             <scope>test</scope>
         </dependency>

diff --git a/beatrix/src/test/java/com/ning/billing/beatrix/integration/BeatrixModule.java b/beatrix/src/test/java/com/ning/billing/beatrix/integration/BeatrixModule.java
index 0e0ea55..c26fac5 100644
--- a/beatrix/src/test/java/com/ning/billing/beatrix/integration/BeatrixModule.java
+++ b/beatrix/src/test/java/com/ning/billing/beatrix/integration/BeatrixModule.java
@@ -29,6 +29,7 @@ import com.google.inject.Inject;
 import com.google.inject.Injector;
 import com.ning.billing.account.api.AccountService;
 import com.ning.billing.account.glue.AccountModule;
+import com.ning.billing.analytics.setup.AnalyticsModule;
 import com.ning.billing.beatrix.integration.overdue.IntegrationTestOverdueModule;
 import com.ning.billing.beatrix.lifecycle.DefaultLifecycle;
 import com.ning.billing.beatrix.lifecycle.Lifecycle;
@@ -96,6 +97,7 @@ public class BeatrixModule extends AbstractModule {
         install(new TagStoreModule());
         install(new CustomFieldModule());
         install(new AccountModule());
+        install(new AnalyticsModule());
         install(new CatalogModule());
         install(new DefaultEntitlementModule());
         install(new DefaultInvoiceModule());

diff --git a/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestAnalytics.java b/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestAnalytics.java
new file mode 100644
index 0000000..9dadb9e
--- /dev/null
+++ b/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestAnalytics.java
@@ -0,0 +1,322 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.billing.beatrix.integration;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Guice;
+import org.testng.annotations.Test;
+
+import com.ning.billing.account.api.Account;
+import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.account.api.AccountData;
+import com.ning.billing.account.api.MutableAccountData;
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessAccountTag;
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
+import com.ning.billing.analytics.utils.Rounder;
+import com.ning.billing.catalog.api.BillingPeriod;
+import com.ning.billing.catalog.api.CatalogApiException;
+import com.ning.billing.catalog.api.PhaseType;
+import com.ning.billing.catalog.api.PlanPhaseSpecifier;
+import com.ning.billing.catalog.api.PriceListSet;
+import com.ning.billing.catalog.api.ProductCategory;
+import com.ning.billing.entitlement.api.user.EntitlementUserApiException;
+import com.ning.billing.entitlement.api.user.Subscription;
+import com.ning.billing.entitlement.api.user.SubscriptionBundle;
+import com.ning.billing.util.api.TagApiException;
+import com.ning.billing.util.api.TagDefinitionApiException;
+import com.ning.billing.util.dao.ObjectType;
+import com.ning.billing.util.tag.TagDefinition;
+
+@Guice(modules = BeatrixModule.class)
+public class TestAnalytics extends TestIntegrationBase {
+    @BeforeMethod(groups = "slow")
+    public void setUpAnalyticsHandler() throws Exception {
+        busService.getBus().register(analyticsListener);
+    }
+
+    @AfterMethod(groups = "slow")
+    public void tearDownAnalyticsHandler() throws Exception {
+        busService.getBus().unregister(analyticsListener);
+    }
+
+    @Test(groups = "slow")
+    public void testAnalyticsEvents() throws Exception {
+        // Create an account
+        final Account account = verifyAccountCreation();
+
+        // Update some fields
+        verifyAccountUpdate(account);
+
+        // Add a tag
+        verifyAddTagToAccount(account);
+
+        // Create a bundle
+        final SubscriptionBundle bundle = verifyFirstBundle(account);
+
+        // Add a subscription
+        final Subscription subscription = verifyFirstSubscription(account, bundle);
+
+        // Upgrade the subscription
+        verifyChangePlan(account, bundle, subscription);
+    }
+
+    private Account verifyAccountCreation() throws Exception {
+        final AccountData accountData = getAccountData(1);
+
+        // Verify BAC is empty
+        Assert.assertNull(analyticsUserApi.getAccountByKey(accountData.getExternalKey()));
+
+        // Create an account
+        final Account account = createAccountWithPaymentMethod(accountData);
+        Assert.assertNotNull(account);
+
+        waitALittle();
+
+        // Verify Analytics got the account creation event
+        final BusinessAccount businessAccount = analyticsUserApi.getAccountByKey(account.getExternalKey());
+        Assert.assertNotNull(businessAccount);
+        // No balance yet
+        Assert.assertEquals(businessAccount.getBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+        Assert.assertEquals(businessAccount.getKey(), account.getExternalKey());
+        // No invoice yet
+        Assert.assertNull(businessAccount.getLastInvoiceDate());
+        // No payment yet
+        Assert.assertNull(businessAccount.getLastPaymentStatus());
+        Assert.assertEquals(businessAccount.getName(), account.getName());
+        // No invoice balance yet
+        Assert.assertEquals(businessAccount.getTotalInvoiceBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+        // TODO - payment fields
+        //Assert.assertNotNull(businessAccount.getBillingAddressCountry());
+        //Assert.assertNotNull(businessAccount.getCreditCardType());
+        //Assert.assertNotNull(businessAccount.getPaymentMethod());
+
+        // The account shouldn't have any invoice yet
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).size(), 0);
+
+        return account;
+    }
+
+    private void verifyAccountUpdate(final Account account) throws InterruptedException {
+        final MutableAccountData mutableAccountData = account.toMutableAccountData();
+
+        mutableAccountData.setName(UUID.randomUUID().toString().substring(0, 20));
+
+        try {
+            accountUserApi.updateAccount(account.getId(), mutableAccountData, context);
+        } catch (AccountApiException e) {
+            Assert.fail("Unable to update account", e);
+        }
+
+        waitALittle();
+
+        // Verify Analytics got the account update event
+        final BusinessAccount businessAccount = analyticsUserApi.getAccountByKey(mutableAccountData.getExternalKey());
+        Assert.assertNotNull(businessAccount);
+        // No balance yet
+        Assert.assertEquals(businessAccount.getBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+        Assert.assertEquals(businessAccount.getKey(), mutableAccountData.getExternalKey());
+        // No invoice yet
+        Assert.assertNull(businessAccount.getLastInvoiceDate());
+        // No payment yet
+        Assert.assertNull(businessAccount.getLastPaymentStatus());
+        Assert.assertEquals(businessAccount.getName(), mutableAccountData.getName());
+        // No invoice balance yet
+        Assert.assertEquals(businessAccount.getTotalInvoiceBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+        // TODO - payment fields
+        //Assert.assertNotNull(businessAccount.getBillingAddressCountry());
+        //Assert.assertNotNull(businessAccount.getCreditCardType());
+        //Assert.assertNotNull(businessAccount.getPaymentMethod());
+
+        // The account should still not have any invoice
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).size(), 0);
+    }
+
+    private void verifyAddTagToAccount(final Account account) throws TagDefinitionApiException, TagApiException, InterruptedException {
+        Assert.assertEquals(analyticsUserApi.getTagsForAccount(account.getExternalKey()).size(), 0);
+
+        final TagDefinition tagDefinition = tagUserApi.create(UUID.randomUUID().toString().substring(0, 10), UUID.randomUUID().toString(), context);
+        tagUserApi.addTag(account.getId(), ObjectType.ACCOUNT, tagDefinition, context);
+
+        waitALittle();
+
+        final List<BusinessAccountTag> tagsForAccount = analyticsUserApi.getTagsForAccount(account.getExternalKey());
+        Assert.assertEquals(tagsForAccount.size(), 1);
+        Assert.assertEquals(tagsForAccount.get(0).getName(), tagDefinition.getName());
+    }
+
+    private SubscriptionBundle verifyFirstBundle(final Account account) throws EntitlementUserApiException, InterruptedException {
+        // Add a bundle
+        final SubscriptionBundle bundle = entitlementUserApi.createBundleForAccount(account.getId(), UUID.randomUUID().toString(), context);
+        Assert.assertNotNull(bundle);
+
+        waitALittle();
+
+        // Verify BST is still empty since no subscription has been added yet
+        Assert.assertEquals(analyticsUserApi.getTransitionsForBundle(bundle.getKey()).size(), 0);
+
+        // The account should still not have any invoice
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).size(), 0);
+
+        return bundle;
+    }
+
+    private Subscription verifyFirstSubscription(final Account account, final SubscriptionBundle bundle) throws EntitlementUserApiException, InterruptedException, CatalogApiException {
+        // Add a subscription
+        final String productName = "Shotgun";
+        final BillingPeriod term = BillingPeriod.MONTHLY;
+        final String planSetName = PriceListSet.DEFAULT_PRICELIST_NAME;
+        final PlanPhaseSpecifier phaseSpecifier = new PlanPhaseSpecifier(productName, ProductCategory.BASE, term, planSetName, null);
+        final Subscription subscription = entitlementUserApi.createSubscription(bundle.getId(), phaseSpecifier, null, context);
+
+        waitALittle();
+
+        // BST should have one transition
+        final List<BusinessSubscriptionTransition> transitions = analyticsUserApi.getTransitionsForBundle(bundle.getKey());
+        Assert.assertEquals(transitions.size(), 1);
+        final BusinessSubscriptionTransition transition = transitions.get(0);
+        Assert.assertEquals(transition.getExternalKey(), bundle.getKey());
+        Assert.assertEquals(transition.getAccountKey(), account.getExternalKey());
+        Assert.assertEquals(transition.getEvent().getCategory(), phaseSpecifier.getProductCategory());
+        Assert.assertEquals(transition.getEvent().getEventType(), BusinessSubscriptionEvent.EventType.ADD);
+
+        // This is the first transition
+        Assert.assertNull(transition.getPreviousSubscription());
+
+        Assert.assertEquals(transition.getNextSubscription().getBillingPeriod(), subscription.getCurrentPhase().getBillingPeriod().toString());
+        Assert.assertEquals(transition.getNextSubscription().getBundleId(), subscription.getBundleId());
+        Assert.assertEquals(transition.getNextSubscription().getCurrency(), account.getCurrency().toString());
+        Assert.assertEquals(transition.getNextSubscription().getPhase(), subscription.getCurrentPhase().getPhaseType().toString());
+        // Trial: fixed price of zero
+        Assert.assertEquals(transition.getNextSubscription().getPrice().doubleValue(), subscription.getCurrentPhase().getFixedPrice().getPrice(account.getCurrency()).doubleValue());
+        Assert.assertEquals(transition.getNextSubscription().getPriceList(), subscription.getCurrentPriceList().getName());
+        Assert.assertEquals(transition.getNextSubscription().getProductCategory(), subscription.getCurrentPlan().getProduct().getCategory());
+        Assert.assertEquals(transition.getNextSubscription().getProductName(), subscription.getCurrentPlan().getProduct().getName());
+        Assert.assertEquals(transition.getNextSubscription().getProductType(), subscription.getCurrentPlan().getProduct().getCatalogName());
+        Assert.assertEquals(transition.getNextSubscription().getSlug(), subscription.getCurrentPhase().getName());
+        Assert.assertEquals(transition.getNextSubscription().getStartDate(), subscription.getStartDate());
+        Assert.assertEquals(transition.getNextSubscription().getState(), subscription.getState());
+        Assert.assertEquals(transition.getNextSubscription().getSubscriptionId(), subscription.getId());
+
+        // Make sure the account balance is still zero
+        final BusinessAccount businessAccount = analyticsUserApi.getAccountByKey(account.getExternalKey());
+        Assert.assertEquals(businessAccount.getBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+        Assert.assertEquals(businessAccount.getTotalInvoiceBalance().doubleValue(), Rounder.round(BigDecimal.ZERO));
+
+        // The account should have one invoice for the trial phase
+        final List<BusinessInvoice> invoices = analyticsUserApi.getInvoicesForAccount(account.getExternalKey());
+        Assert.assertEquals(invoices.size(), 1);
+        final BusinessInvoice invoice = invoices.get(0);
+        Assert.assertEquals(invoice.getBalance().doubleValue(), 0.0);
+        Assert.assertEquals(invoice.getAmountCharged().doubleValue(), 0.0);
+        Assert.assertEquals(invoice.getAmountCredited().doubleValue(), 0.0);
+        Assert.assertEquals(invoice.getAmountPaid().doubleValue(), 0.0);
+        Assert.assertEquals(invoice.getCurrency(), account.getCurrency());
+
+        // The invoice should have a single item associated to it
+        final List<BusinessInvoiceItem> invoiceItems = analyticsUserApi.getInvoiceItemsForInvoice(invoice.getInvoiceId());
+        Assert.assertEquals(invoiceItems.size(), 1);
+        final BusinessInvoiceItem invoiceItem = invoiceItems.get(0);
+        Assert.assertEquals(invoiceItem.getAmount().doubleValue(), 0.0);
+        // No billing period for the trial item
+        Assert.assertEquals(invoiceItem.getBillingPeriod(), subscription.getCurrentPhase().getBillingPeriod().toString());
+        Assert.assertEquals(invoiceItem.getCurrency(), account.getCurrency());
+        // The subscription end date is null (evergreen)
+        Assert.assertEquals(invoiceItem.getEndDate(), subscription.getStartDate().plus(subscription.getCurrentPhase().getDuration().toJodaPeriod()));
+        Assert.assertEquals(invoiceItem.getExternalKey(), bundle.getKey());
+        Assert.assertEquals(invoiceItem.getInvoiceId(), invoice.getInvoiceId());
+        Assert.assertEquals(invoiceItem.getItemType(), "FIXED");
+        Assert.assertEquals(invoiceItem.getPhase(), subscription.getCurrentPhase().getPhaseType().toString());
+        Assert.assertEquals(invoiceItem.getProductCategory(), subscription.getCurrentPlan().getProduct().getCategory().toString());
+        Assert.assertEquals(invoiceItem.getProductName(), subscription.getCurrentPlan().getProduct().getName());
+        Assert.assertEquals(invoiceItem.getProductType(), subscription.getCurrentPlan().getProduct().getCatalogName());
+        Assert.assertEquals(invoiceItem.getSlug(), subscription.getCurrentPhase().getName());
+        Assert.assertEquals(invoiceItem.getStartDate(), subscription.getStartDate());
+
+        return subscription;
+    }
+
+    private void verifyChangePlan(final Account account, final SubscriptionBundle bundle, final Subscription subscription) throws EntitlementUserApiException, InterruptedException {
+        final String newProductName = "Assault-Rifle";
+        final BillingPeriod newTerm = BillingPeriod.MONTHLY;
+        final String newPlanSetName = PriceListSet.DEFAULT_PRICELIST_NAME;
+        final DateTime requestedDate = clock.getUTCNow();
+        Assert.assertTrue(subscription.changePlan(newProductName, newTerm, newPlanSetName, requestedDate, context));
+
+        waitALittle();
+
+        // BST should have two transitions
+        final List<BusinessSubscriptionTransition> transitions = analyticsUserApi.getTransitionsForBundle(bundle.getKey());
+        Assert.assertEquals(transitions.size(), 2);
+        final BusinessSubscriptionTransition previousTransition = transitions.get(0);
+        final BusinessSubscriptionTransition transition = transitions.get(1);
+        Assert.assertEquals(transition.getExternalKey(), bundle.getKey());
+        Assert.assertEquals(transition.getAccountKey(), account.getExternalKey());
+        Assert.assertEquals(transition.getEvent().getCategory(), ProductCategory.BASE);
+        Assert.assertEquals(transition.getEvent().getEventType(), BusinessSubscriptionEvent.EventType.CHANGE);
+
+        // Verify the previous subscription matches
+        Assert.assertNull(previousTransition.getPreviousSubscription());
+        Assert.assertEquals(previousTransition.getNextSubscription(), transition.getPreviousSubscription());
+
+        // Verify the next subscription
+        // No billing period for the trial phase
+        Assert.assertEquals(transition.getNextSubscription().getBillingPeriod(), BillingPeriod.NO_BILLING_PERIOD.toString());
+        Assert.assertEquals(transition.getNextSubscription().getBundleId(), subscription.getBundleId());
+        Assert.assertEquals(transition.getNextSubscription().getCurrency(), account.getCurrency().toString());
+        Assert.assertEquals(transition.getNextSubscription().getPhase(), PhaseType.TRIAL.toString());
+        // We're still in trial
+        Assert.assertEquals(transition.getNextSubscription().getPrice().doubleValue(), 0.0);
+        Assert.assertEquals(transition.getNextSubscription().getPriceList(), newPlanSetName);
+        Assert.assertEquals(transition.getNextSubscription().getProductCategory(), ProductCategory.BASE);
+        Assert.assertEquals(transition.getNextSubscription().getProductName(), newProductName);
+        Assert.assertEquals(transition.getNextSubscription().getProductType(), subscription.getCurrentPlan().getProduct().getCatalogName());
+        Assert.assertEquals(transition.getNextSubscription().getSlug(), subscription.getCurrentPhase().getName());
+        Assert.assertEquals(transition.getNextSubscription().getStartDate(), requestedDate);
+        Assert.assertEquals(transition.getNextSubscription().getState(), Subscription.SubscriptionState.ACTIVE);
+        // It's still the same subscription
+        Assert.assertEquals(transition.getNextSubscription().getSubscriptionId(), subscription.getId());
+
+        // The account should have two invoices for the trial phase of both subscriptions
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).size(), 2);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(0).getBalance().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(0).getAmountCharged().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(0).getAmountCredited().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(0).getAmountPaid().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(0).getCurrency(), account.getCurrency());
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(1).getBalance().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(1).getAmountCharged().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(1).getAmountCredited().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(1).getAmountPaid().doubleValue(), 0.0);
+        Assert.assertEquals(analyticsUserApi.getInvoicesForAccount(account.getExternalKey()).get(1).getCurrency(), account.getCurrency());
+    }
+
+    private void waitALittle() throws InterruptedException {
+        // We especially need to wait for entitlement events
+        Thread.sleep(2000);
+    }
+}

diff --git a/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestIntegrationBase.java b/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestIntegrationBase.java
index 43ccf5d..d17b0b2 100644
--- a/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestIntegrationBase.java
+++ b/beatrix/src/test/java/com/ning/billing/beatrix/integration/TestIntegrationBase.java
@@ -37,6 +37,8 @@ import com.ning.billing.account.api.Account;
 import com.ning.billing.account.api.AccountData;
 import com.ning.billing.account.api.AccountService;
 import com.ning.billing.account.api.AccountUserApi;
+import com.ning.billing.analytics.AnalyticsListener;
+import com.ning.billing.analytics.api.user.DefaultAnalyticsUserApi;
 import com.ning.billing.api.TestApiListener;
 import com.ning.billing.api.TestListenerStatus;
 import com.ning.billing.beatrix.lifecycle.Lifecycle;
@@ -56,6 +58,7 @@ import com.ning.billing.invoice.model.InvoicingConfiguration;
 import com.ning.billing.junction.plumbing.api.BlockingSubscription;
 import com.ning.billing.payment.api.PaymentApi;
 import com.ning.billing.payment.api.PaymentMethodPlugin;
+import com.ning.billing.util.api.TagUserApi;
 import com.ning.billing.util.bus.BusService;
 import com.ning.billing.util.callcontext.CallContext;
 import com.ning.billing.util.callcontext.CallOrigin;
@@ -124,6 +127,15 @@ public class TestIntegrationBase implements TestListenerStatus {
     @Inject
     protected AccountUserApi accountUserApi;
 
+    @Inject
+    protected DefaultAnalyticsUserApi analyticsUserApi;
+
+    @Inject
+    protected TagUserApi tagUserApi;
+
+    @Inject
+    protected AnalyticsListener analyticsListener;
+
     protected TestApiListener busHandler;
 
 
@@ -152,6 +164,7 @@ public class TestIntegrationBase implements TestListenerStatus {
 
     protected void setupMySQL() throws IOException {
         final String accountDdl = IOUtils.toString(TestIntegration.class.getResourceAsStream("/com/ning/billing/account/ddl.sql"));
+        final String analyticsDdl = IOUtils.toString(TestIntegration.class.getResourceAsStream("/com/ning/billing/analytics/ddl.sql"));
         final String entitlementDdl = IOUtils.toString(TestIntegration.class.getResourceAsStream("/com/ning/billing/entitlement/ddl.sql"));
         final String invoiceDdl = IOUtils.toString(TestIntegration.class.getResourceAsStream("/com/ning/billing/invoice/ddl.sql"));
         final String paymentDdl = IOUtils.toString(TestIntegration.class.getResourceAsStream("/com/ning/billing/payment/ddl.sql"));
@@ -161,6 +174,7 @@ public class TestIntegrationBase implements TestListenerStatus {
         helper.startMysql();
 
         helper.initDb(accountDdl);
+        helper.initDb(analyticsDdl);
         helper.initDb(entitlementDdl);
         helper.initDb(invoiceDdl);
         helper.initDb(paymentDdl);

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/migration/DefaultEntitlementMigrationApi.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/migration/DefaultEntitlementMigrationApi.java
index ee4bc32..e0f2860 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/migration/DefaultEntitlementMigrationApi.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/migration/DefaultEntitlementMigrationApi.java
@@ -91,7 +91,7 @@ public class DefaultEntitlementMigrationApi implements EntitlementMigrationApi {
 
 
             final List<EntitlementSubscriptionMigration> sortedSubscriptions = Lists.newArrayList(curBundle.getSubscriptions());
-            // Make sure we have first mpp or legacy, then addon and for each category order by CED
+            // Make sure we have first BASE or STANDALONE, then ADDON and for each category order by CED
             Collections.sort(sortedSubscriptions, new Comparator<EntitlementSubscriptionMigration>() {
                 @Override
                 public int compare(final EntitlementSubscriptionMigration o1,
@@ -99,14 +99,10 @@ public class DefaultEntitlementMigrationApi implements EntitlementMigrationApi {
                     if (o1.getCategory().equals(o2.getCategory())) {
                         return o1.getSubscriptionCases()[0].getEffectiveDate().compareTo(o2.getSubscriptionCases()[0].getEffectiveDate());
                     } else {
-                        if (o1.getCategory().equals("mpp")) {
+                    	if (!o1.getCategory().name().equalsIgnoreCase("ADD_ON")) {
                             return -1;
-                        } else if (o2.getCategory().equals("mpp")) {
-                            return 1;
-                        } else if (o1.getCategory().equals("legacy")) {
-                            return -1;
-                        } else if (o2.getCategory().equals("legacy")) {
-                            return 1;
+                        } else if (o1.getCategory().name().equalsIgnoreCase("ADD_ON")) {
+                            return 1;                    
                         } else {
                             return 0;
                         }