killbill-memoizeit

diff --git a/account/src/main/resources/com/ning/billing/account/ddl.sql b/account/src/main/resources/com/ning/billing/account/ddl.sql
index 2cf54d1..7ba749e 100644
--- a/account/src/main/resources/com/ning/billing/account/ddl.sql
+++ b/account/src/main/resources/com/ning/billing/account/ddl.sql
@@ -75,7 +75,6 @@ CREATE TABLE account_emails (
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
 CREATE UNIQUE INDEX account_email_id ON account_emails(id);
-CREATE INDEX account_email_account_id ON account_emails(account_id);
 CREATE UNIQUE INDEX account_email_account_id_email ON account_emails(account_id, email);
 
 DROP TABLE IF EXISTS account_email_history;

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 0877ae4..7f69fef 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/BusinessAccountRecorder.java
@@ -32,7 +32,8 @@ 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.BusinessAccountDao;
+import com.ning.billing.analytics.dao.BusinessAccountSqlDao;
+import com.ning.billing.analytics.model.BusinessAccount;
 import com.ning.billing.invoice.api.Invoice;
 import com.ning.billing.invoice.api.InvoiceUserApi;
 import com.ning.billing.payment.api.Payment;
@@ -46,17 +47,17 @@ import com.ning.billing.util.tag.Tag;
 public class BusinessAccountRecorder {
     private static final Logger log = LoggerFactory.getLogger(BusinessAccountRecorder.class);
 
-    private final BusinessAccountDao dao;
+    private final BusinessAccountSqlDao sqlDao;
     private final AccountUserApi accountApi;
     private final InvoiceUserApi invoiceUserApi;
     private final PaymentApi paymentApi;
     private final TagUserApi tagUserApi;
 
     @Inject
-    public BusinessAccountRecorder(final BusinessAccountDao dao, final AccountUserApi accountApi,
+    public BusinessAccountRecorder(final BusinessAccountSqlDao sqlDao, final AccountUserApi accountApi,
                                    final InvoiceUserApi invoiceUserApi, final PaymentApi paymentApi,
                                    final TagUserApi tagUserApi) {
-        this.dao = dao;
+        this.sqlDao = sqlDao;
         this.accountApi = accountApi;
         this.invoiceUserApi = invoiceUserApi;
         this.paymentApi = paymentApi;
@@ -71,7 +72,7 @@ public class BusinessAccountRecorder {
             final BusinessAccount bac = createBusinessAccountFromAccount(account, new ArrayList<Tag>(tags.values()));
 
             log.info("ACCOUNT CREATION " + bac);
-            dao.createAccount(bac);
+            sqlDao.createAccount(bac);
         } catch (AccountApiException e) {
             log.warn("Error encountered creating BusinessAccount", e);
         }
@@ -117,15 +118,15 @@ public class BusinessAccountRecorder {
                 return;
             }
 
-            BusinessAccount bac = dao.getAccount(account.getExternalKey());
+            BusinessAccount bac = sqlDao.getAccount(account.getExternalKey());
             if (bac == null) {
                 bac = createBusinessAccountFromAccount(account, new ArrayList<Tag>(tags.values()));
                 log.info("ACCOUNT CREATION " + bac);
-                dao.createAccount(bac);
+                sqlDao.createAccount(bac);
             } else {
                 updateBusinessAccountFromAccount(account, bac);
                 log.info("ACCOUNT UPDATE " + bac);
-                dao.saveAccount(bac);
+                sqlDao.saveAccount(bac);
             }
         } catch (AccountApiException e) {
             log.warn("Error encountered creating BusinessAccount", e);
@@ -136,8 +137,8 @@ public class BusinessAccountRecorder {
     private BusinessAccount createBusinessAccountFromAccount(final Account account, final List<Tag> tags) {
         final BusinessAccount bac = new BusinessAccount(
                 account.getExternalKey(),
+                account.getName(),
                 invoiceUserApi.getAccountBalance(account.getId()),
-                tags,
                 // These fields will be updated below
                 null,
                 null,

diff --git a/analytics/src/main/java/com/ning/billing/analytics/BusinessSubscriptionTransitionRecorder.java b/analytics/src/main/java/com/ning/billing/analytics/BusinessSubscriptionTransitionRecorder.java
index 80db909..04e3547 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/BusinessSubscriptionTransitionRecorder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/BusinessSubscriptionTransitionRecorder.java
@@ -17,7 +17,6 @@
 package com.ning.billing.analytics;
 
 import java.util.List;
-import java.util.UUID;
 
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
@@ -27,7 +26,10 @@ import com.google.inject.Inject;
 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.BusinessSubscriptionTransitionDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDao;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.catalog.api.CatalogService;
 import com.ning.billing.catalog.api.Currency;
 import com.ning.billing.entitlement.api.user.EntitlementUserApi;
@@ -38,14 +40,14 @@ import com.ning.billing.entitlement.api.user.SubscriptionEvent;
 public class BusinessSubscriptionTransitionRecorder {
     private static final Logger log = LoggerFactory.getLogger(BusinessSubscriptionTransitionRecorder.class);
 
-    private final BusinessSubscriptionTransitionDao dao;
+    private final BusinessSubscriptionTransitionSqlDao sqlDao;
     private final EntitlementUserApi entitlementApi;
     private final AccountUserApi accountApi;
     private final CatalogService catalogService;
 
     @Inject
-    public BusinessSubscriptionTransitionRecorder(final BusinessSubscriptionTransitionDao dao, final CatalogService catalogService, final EntitlementUserApi entitlementApi, final AccountUserApi accountApi) {
-        this.dao = dao;
+    public BusinessSubscriptionTransitionRecorder(final BusinessSubscriptionTransitionSqlDao sqlDao, final CatalogService catalogService, final EntitlementUserApi entitlementApi, final AccountUserApi accountApi) {
+        this.sqlDao = sqlDao;
         this.catalogService = catalogService;
         this.entitlementApi = entitlementApi;
         this.accountApi = accountApi;
@@ -84,13 +86,13 @@ public class BusinessSubscriptionTransitionRecorder {
     void recordTransition(final BusinessSubscriptionEvent event, final SubscriptionEvent transition)
             throws AccountApiException, EntitlementUserApiException {
         Currency currency = null;
-        String transitionKey = null;
+        String externalKey = null;
         String accountKey = null;
 
         // Retrieve key and currency via the bundle
         final SubscriptionBundle bundle = entitlementApi.getBundleFromId(transition.getBundleId());
         if (bundle != null) {
-            transitionKey = bundle.getKey();
+            externalKey = bundle.getKey();
 
             final Account account = accountApi.getAccountById(bundle.getAccountId());
             if (account != null) {
@@ -104,7 +106,7 @@ public class BusinessSubscriptionTransitionRecorder {
         DateTime previousEffectiveTransitionTime = null;
         // For creation events, the prev subscription will always be null
         if (event.getEventType() != BusinessSubscriptionEvent.EventType.ADD) {
-            final List<BusinessSubscriptionTransition> transitions = dao.getTransitions(transitionKey);
+            final List<BusinessSubscriptionTransition> transitions = sqlDao.getTransitions(externalKey);
             if (transitions != null && transitions.size() > 0) {
                 final BusinessSubscriptionTransition lastTransition = transitions.get(transitions.size() - 1);
                 if (lastTransition != null && lastTransition.getNextSubscription() != null) {
@@ -129,14 +131,14 @@ public class BusinessSubscriptionTransitionRecorder {
             nextSubscription = new BusinessSubscription(transition.getNextPriceList(), transition.getNextPlan(), transition.getNextPhase(), currency, transition.getEffectiveTransitionTime(), transition.getNextState(), transition.getSubscriptionId(), transition.getBundleId(), catalogService.getFullCatalog());
         }
 
-        record(transition.getId(), transitionKey, accountKey, transition.getRequestedTransitionTime(), event, prevSubscription, nextSubscription);
+        record(transition.getTotalOrdering(), externalKey, accountKey, transition.getRequestedTransitionTime(), event, prevSubscription, nextSubscription);
     }
 
     // Public for internal reasons
-    public void record(final UUID id, final String key, final String accountKey, final DateTime requestedDateTime, final BusinessSubscriptionEvent event, final BusinessSubscription prevSubscription, final BusinessSubscription nextSubscription) {
+    public void record(final Long totalOrdering, final String externalKey, final String accountKey, final DateTime requestedDateTime, final BusinessSubscriptionEvent event, final BusinessSubscription prevSubscription, final BusinessSubscription nextSubscription) {
         final BusinessSubscriptionTransition transition = new BusinessSubscriptionTransition(
-                id,
-                key,
+                totalOrdering,
+                externalKey,
                 accountKey,
                 requestedDateTime,
                 event,
@@ -145,6 +147,6 @@ public class BusinessSubscriptionTransitionRecorder {
         );
 
         log.info(transition.getEvent() + " " + transition);
-        dao.createTransition(transition);
+        sqlDao.createTransition(transition);
     }
 }

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountBinder.java
index 234e543..223068e 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountBinder.java
@@ -30,16 +30,13 @@ import org.skife.jdbi.v2.sqlobject.Binder;
 import org.skife.jdbi.v2.sqlobject.BinderFactory;
 import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
 
-import com.google.common.base.Joiner;
-import com.ning.billing.analytics.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessAccount;
 
 @BindingAnnotation(BusinessAccountBinder.BacBinderFactory.class)
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.PARAMETER})
 public @interface BusinessAccountBinder {
     public static class BacBinderFactory implements BinderFactory {
-        private final Joiner joiner = Joiner.on(";").skipNulls();
-
         public Binder build(final Annotation annotation) {
             return new Binder<BusinessAccountBinder, BusinessAccount>() {
                 public void bind(final SQLStatement q, final BusinessAccountBinder bind, final BusinessAccount account) {
@@ -54,7 +51,7 @@ public @interface BusinessAccountBinder {
 
                     q.bind("account_key", account.getKey());
                     q.bind("balance", account.getRoundedBalance());
-                    q.bind("tags", joiner.join(account.getTags()));
+                    q.bind("name", account.getName());
                     if (account.getLastInvoiceDate() != null) {
                         q.bind("last_invoice_date", account.getLastInvoiceDate().getMillis());
                     } else {

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldMapper.java
new file mode 100644
index 0000000..7cb7260
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountFieldMapper.java
@@ -0,0 +1,32 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessAccountField;
+
+public class BusinessAccountFieldMapper implements ResultSetMapper<BusinessAccountField> {
+    @Override
+    public BusinessAccountField map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessAccountField(r.getString(1), r.getString(2), r.getString(3));
+    }
+}

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 366bb3b..a28864b 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
@@ -19,54 +19,21 @@ package com.ning.billing.analytics.dao;
 import java.math.BigDecimal;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-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.StatementContext;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-import com.ning.billing.analytics.BusinessAccount;
-import com.ning.billing.util.tag.Tag;
+import com.ning.billing.analytics.model.BusinessAccount;
 
 public class BusinessAccountMapper implements ResultSetMapper<BusinessAccount> {
-    private final Splitter splitter = Splitter.on(";").trimResults().omitEmptyStrings();
-
     @Override
     public BusinessAccount map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
-        final List<String> tagNames = new ArrayList<String>();
-        Iterables.addAll(tagNames, splitter.split(r.getString(5)));
-
-        final List<Tag> tags = new ArrayList<Tag>();
-        for (final String tagName : tagNames) {
-            tags.add(new Tag() {
-                private final UUID id = UUID.randomUUID();
-
-                @Override
-                public String getTagDefinitionName() {
-                    return tagName;
-                }
-
-                @Override
-                public UUID getId() {
-                    return id;
-                }
-
-                @Override
-                public String toString() {
-                    return tagName;
-                }
-            });
-        }
-
         final BusinessAccount account = new BusinessAccount(
                 r.getString(1),
+                r.getString(5),
                 BigDecimal.valueOf(r.getDouble(4)),
-                tags,
                 new DateTime(r.getLong(6), DateTimeZone.UTC),
                 BigDecimal.valueOf(r.getDouble(7)),
                 r.getString(8),

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagMapper.java
new file mode 100644
index 0000000..ddad81d
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessAccountTagMapper.java
@@ -0,0 +1,32 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessAccountTag;
+
+public class BusinessAccountTagMapper implements ResultSetMapper<BusinessAccountTag> {
+    @Override
+    public BusinessAccountTag map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessAccountTag(r.getString(1), r.getString(2));
+    }
+}

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
new file mode 100644
index 0000000..ad63e3f
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceBinder.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.lang.annotation.Annotation;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.sql.Types;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.SQLStatement;
+import org.skife.jdbi.v2.sqlobject.Binder;
+import org.skife.jdbi.v2.sqlobject.BinderFactory;
+import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
+
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.analytics.utils.Rounder;
+
+@BindingAnnotation(BusinessInvoiceBinder.BinBinderFactory.class)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.PARAMETER})
+public @interface BusinessInvoiceBinder {
+    public static class BinBinderFactory implements BinderFactory {
+        public Binder build(final Annotation annotation) {
+            return new Binder<BusinessInvoiceBinder, BusinessInvoice>() {
+                public void bind(final SQLStatement q, final BusinessInvoiceBinder bind, final BusinessInvoice invoice) {
+                    q.bind("invoice_id", invoice.getInvoiceId().toString());
+
+                    final DateTime dateTimeNow = new DateTime(DateTimeZone.UTC);
+                    if (invoice.getCreatedDate() != null) {
+                        q.bind("created_date", invoice.getCreatedDate().getMillis());
+                    } else {
+                        q.bind("created_date", dateTimeNow.getMillis());
+                    }
+
+                    if (invoice.getUpdatedDate() != null) {
+                        q.bind("updated_date", invoice.getUpdatedDate().getMillis());
+                    } else {
+                        q.bind("updated_date", dateTimeNow.getMillis());
+                    }
+
+                    q.bind("account_key", invoice.getAccountKey());
+
+                    if (invoice.getInvoiceDate() != null) {
+                        q.bind("invoice_date", invoice.getInvoiceDate());
+                    } else {
+                        q.bindNull("invoice_date", Types.BIGINT);
+                    }
+
+                    if (invoice.getTargetDate() != null) {
+                        q.bind("target_date", invoice.getTargetDate());
+                    } else {
+                        q.bindNull("target_date", Types.BIGINT);
+                    }
+
+                    q.bind("currency", invoice.getCurrency().toString());
+                    q.bind("balance", Rounder.round(invoice.getBalance()));
+                    q.bind("amount_paid", Rounder.round(invoice.getAmountPaid()));
+                    q.bind("amount_charged", Rounder.round(invoice.getAmountCharged()));
+                    q.bind("amount_credited", Rounder.round(invoice.getAmountCredited()));
+                }
+            };
+        }
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldMapper.java
new file mode 100644
index 0000000..d75dffc
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceFieldMapper.java
@@ -0,0 +1,33 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoiceField;
+
+public class BusinessInvoiceFieldMapper implements ResultSetMapper<BusinessInvoiceField> {
+    @Override
+    public BusinessInvoiceField map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessInvoiceField(UUID.fromString(r.getString(1)), r.getString(2), r.getString(3));
+    }
+}

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
new file mode 100644
index 0000000..b171a5c
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemBinder.java
@@ -0,0 +1,87 @@
+/*
+ * 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.lang.annotation.Annotation;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.sql.Types;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.SQLStatement;
+import org.skife.jdbi.v2.sqlobject.Binder;
+import org.skife.jdbi.v2.sqlobject.BinderFactory;
+import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
+
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.analytics.utils.Rounder;
+
+@BindingAnnotation(BusinessInvoiceItemBinder.BiiBinderFactory.class)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.PARAMETER})
+public @interface BusinessInvoiceItemBinder {
+    public static class BiiBinderFactory implements BinderFactory {
+        public Binder build(final Annotation annotation) {
+            return new Binder<BusinessInvoiceItemBinder, BusinessInvoiceItem>() {
+                public void bind(final SQLStatement q, final BusinessInvoiceItemBinder bind, final BusinessInvoiceItem invoiceItem) {
+                    q.bind("item_id", invoiceItem.getItemId().toString());
+
+                    final DateTime dateTimeNow = new DateTime(DateTimeZone.UTC);
+                    if (invoiceItem.getCreatedDate() != null) {
+                        q.bind("created_date", invoiceItem.getCreatedDate().getMillis());
+                    } else {
+                        q.bind("created_date", dateTimeNow.getMillis());
+                    }
+
+                    if (invoiceItem.getUpdatedDate() != null) {
+                        q.bind("updated_date", invoiceItem.getUpdatedDate().getMillis());
+                    } else {
+                        q.bind("updated_date", dateTimeNow.getMillis());
+                    }
+
+                    q.bind("invoice_id", invoiceItem.getInvoiceId().toString());
+                    q.bind("item_type", invoiceItem.getItemType());
+                    q.bind("external_key", invoiceItem.getExternalKey());
+                    q.bind("product_name", invoiceItem.getProductName());
+                    q.bind("product_type", invoiceItem.getProductType());
+                    q.bind("product_category", invoiceItem.getProductCategory());
+                    q.bind("slug", invoiceItem.getSlug());
+                    q.bind("phase", invoiceItem.getPhase());
+                    q.bind("billing_period", invoiceItem.getBillingPeriod());
+
+                    if (invoiceItem.getStartDate() != null) {
+                        q.bind("start_date", invoiceItem.getStartDate());
+                    } else {
+                        q.bindNull("start_date", Types.BIGINT);
+                    }
+
+                    if (invoiceItem.getEndDate() != null) {
+                        q.bind("end_date", invoiceItem.getEndDate());
+                    } else {
+                        q.bindNull("end_date", Types.BIGINT);
+                    }
+
+                    q.bind("amount", Rounder.round(invoiceItem.getAmount()));
+                    q.bind("currency", invoiceItem.getCurrency().toString());
+                }
+            };
+        }
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemMapper.java
new file mode 100644
index 0000000..b74f5b0
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemMapper.java
@@ -0,0 +1,56 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoiceItem;
+import com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoiceItemMapper implements ResultSetMapper<BusinessInvoiceItem> {
+    @Override
+    public BusinessInvoiceItem map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        final UUID itemId = UUID.fromString(r.getString(1));
+        final DateTime createdDate = new DateTime(r.getLong(2), DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(r.getLong(3), DateTimeZone.UTC);
+        final UUID invoiceId = UUID.fromString(r.getString(4));
+        final String itemType = r.getString(5);
+        final String externalKey = r.getString(6);
+        final String productName = r.getString(7);
+        final String productType = r.getString(8);
+        final String productCategory = r.getString(9);
+        final String slug = r.getString(10);
+        final String phase = r.getString(11);
+        final String billingPeriod = r.getString(12);
+        final DateTime startDate = new DateTime(r.getLong(13), DateTimeZone.UTC);
+        final DateTime endDate = new DateTime(r.getLong(14), DateTimeZone.UTC);
+        final BigDecimal amount = BigDecimal.valueOf(r.getDouble(15));
+        final Currency currency = Currency.valueOf(r.getString(16));
+
+        return new BusinessInvoiceItem(amount, billingPeriod, createdDate, currency, endDate, externalKey, invoiceId,
+                                       itemId, itemType, phase, productCategory, productName, productType, slug,
+                                       startDate, updatedDate);
+    }
+}

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
new file mode 100644
index 0000000..6f491d6
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.java
@@ -0,0 +1,52 @@
+/*
+ * 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.BusinessInvoiceItem;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoiceItemMapper.class)
+public interface BusinessInvoiceItemSqlDao {
+    @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/BusinessInvoiceMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceMapper.java
new file mode 100644
index 0000000..93d727a
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceMapper.java
@@ -0,0 +1,50 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoice;
+import com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoiceMapper implements ResultSetMapper<BusinessInvoice> {
+    @Override
+    public BusinessInvoice map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        final UUID invoiceId = UUID.fromString(r.getString(1));
+        final DateTime createdDate = new DateTime(r.getLong(2), DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(r.getLong(3), DateTimeZone.UTC);
+        final String accountKey = r.getString(4);
+        final DateTime invoiceDate = new DateTime(r.getLong(5), DateTimeZone.UTC);
+        final DateTime targetDate = new DateTime(r.getLong(6), DateTimeZone.UTC);
+        final Currency currency = Currency.valueOf(r.getString(7));
+        final BigDecimal balance = BigDecimal.valueOf(r.getDouble(8));
+        final BigDecimal amountPaid = BigDecimal.valueOf(r.getDouble(9));
+        final BigDecimal amountCharged = BigDecimal.valueOf(r.getDouble(10));
+        final BigDecimal amountCredited = BigDecimal.valueOf(r.getDouble(11));
+
+        return new BusinessInvoice(accountKey, amountCharged, amountCredited, amountPaid, balance, createdDate, currency,
+                                   invoiceDate, invoiceId, targetDate, updatedDate);
+    }
+}

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
new file mode 100644
index 0000000..4cc614b
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentBinder.java
@@ -0,0 +1,83 @@
+/*
+ * 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.lang.annotation.Annotation;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.sql.Types;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.SQLStatement;
+import org.skife.jdbi.v2.sqlobject.Binder;
+import org.skife.jdbi.v2.sqlobject.BinderFactory;
+import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
+
+import com.ning.billing.analytics.model.BusinessInvoicePayment;
+import com.ning.billing.analytics.utils.Rounder;
+
+@BindingAnnotation(BusinessInvoicePaymentBinder.BipBinderFactory.class)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.PARAMETER})
+public @interface BusinessInvoicePaymentBinder {
+    public static class BipBinderFactory implements BinderFactory {
+        public Binder build(final Annotation annotation) {
+            return new Binder<BusinessInvoicePaymentBinder, BusinessInvoicePayment>() {
+                public void bind(final SQLStatement q, final BusinessInvoicePaymentBinder bind, final BusinessInvoicePayment invoicePayment) {
+                    q.bind("payment_id", invoicePayment.getPaymentId().toString());
+
+                    final DateTime dateTimeNow = new DateTime(DateTimeZone.UTC);
+                    if (invoicePayment.getCreatedDate() != null) {
+                        q.bind("created_date", invoicePayment.getCreatedDate().getMillis());
+                    } else {
+                        q.bind("created_date", dateTimeNow.getMillis());
+                    }
+
+                    if (invoicePayment.getUpdatedDate() != null) {
+                        q.bind("updated_date", invoicePayment.getUpdatedDate().getMillis());
+                    } else {
+                        q.bind("updated_date", dateTimeNow.getMillis());
+                    }
+
+                    q.bind("attempt_id", invoicePayment.getAttemptId().toString());
+                    q.bind("account_key", invoicePayment.getAccountKey());
+                    q.bind("invoice_id", invoicePayment.getInvoiceId().toString());
+
+                    if (invoicePayment.getEffectiveDate() != null) {
+                        q.bind("effective_date", invoicePayment.getEffectiveDate());
+                    } else {
+                        q.bindNull("effective_date", Types.BIGINT);
+                    }
+
+                    q.bind("amount", Rounder.round(invoicePayment.getAmount()));
+                    q.bind("currency", invoicePayment.getCurrency().toString());
+                    q.bind("payment_error", invoicePayment.getPaymentError());
+                    q.bind("processing_status", invoicePayment.getProcessingStatus());
+                    q.bind("requested_amount", Rounder.round(invoicePayment.getRequestedAmount()));
+                    q.bind("plugin_name", invoicePayment.getPluginName());
+                    q.bind("payment_type", invoicePayment.getPaymentType());
+                    q.bind("payment_method", invoicePayment.getPaymentMethod());
+                    q.bind("card_type", invoicePayment.getCardType());
+                    q.bind("card_country", invoicePayment.getCardCountry());
+                }
+            };
+        }
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldMapper.java
new file mode 100644
index 0000000..ac391e1
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentFieldMapper.java
@@ -0,0 +1,33 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoicePaymentField;
+
+public class BusinessInvoicePaymentFieldMapper implements ResultSetMapper<BusinessInvoicePaymentField> {
+    @Override
+    public BusinessInvoicePaymentField map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessInvoicePaymentField(UUID.fromString(r.getString(1)), r.getString(2), r.getString(3));
+    }
+}

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/BusinessInvoicePaymentMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentMapper.java
new file mode 100644
index 0000000..af80d53
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentMapper.java
@@ -0,0 +1,57 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoicePayment;
+import com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoicePaymentMapper implements ResultSetMapper<BusinessInvoicePayment> {
+    @Override
+    public BusinessInvoicePayment map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        final UUID paymentId = UUID.fromString(r.getString(1));
+        final DateTime createdDate = new DateTime(r.getLong(2), DateTimeZone.UTC);
+        final DateTime updatedDate = new DateTime(r.getLong(3), DateTimeZone.UTC);
+        final UUID attemptId = UUID.fromString(r.getString(4));
+        final String accountKey = r.getString(5);
+        final UUID invoiceId = UUID.fromString(r.getString(6));
+        final DateTime effectiveDate = new DateTime(r.getLong(7), DateTimeZone.UTC);
+        final BigDecimal amount = BigDecimal.valueOf(r.getDouble(8));
+        final Currency currency = Currency.valueOf(r.getString(9));
+        final String paymentError = r.getString(10);
+        final String processingStatus = r.getString(11);
+        final BigDecimal requestedAmount = BigDecimal.valueOf(r.getDouble(12));
+        final String pluginName = r.getString(13);
+        final String paymentType = r.getString(14);
+        final String paymentMethod = r.getString(15);
+        final String cardType = r.getString(16);
+        final String cardCountry = r.getString(17);
+
+        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/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java
new file mode 100644
index 0000000..681378d
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.java
@@ -0,0 +1,52 @@
+/*
+ * 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.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/BusinessInvoicePaymentTagMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagMapper.java
new file mode 100644
index 0000000..c81c532
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoicePaymentTagMapper.java
@@ -0,0 +1,33 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoicePaymentTag;
+
+public class BusinessInvoicePaymentTagMapper implements ResultSetMapper<BusinessInvoicePaymentTag> {
+    @Override
+    public BusinessInvoicePaymentTag map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessInvoicePaymentTag(UUID.fromString(r.getString(1)), r.getString(2));
+    }
+}

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
new file mode 100644
index 0000000..abfa302
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.java
@@ -0,0 +1,49 @@
+/*
+ * 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.BusinessInvoice;
+
+@ExternalizedSqlViaStringTemplate3()
+@RegisterMapper(BusinessInvoiceMapper.class)
+public interface BusinessInvoiceSqlDao {
+    @SqlQuery
+    BusinessInvoice getInvoice(@Bind("invoice_id") final String invoiceId);
+
+    @SqlQuery
+    List<BusinessInvoice> getInvoicesForAccount(@Bind("account_key") final String accountKey);
+
+    @SqlUpdate
+    int createInvoice(@BusinessInvoiceBinder final BusinessInvoice invoice);
+
+    @SqlUpdate
+    int updateInvoice(@BusinessInvoiceBinder final BusinessInvoice invoice);
+
+    @SqlUpdate
+    int deleteInvoice(@Bind("invoice_id") final String invoiceId);
+
+    @SqlUpdate
+    void test();
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagMapper.java
new file mode 100644
index 0000000..c812143
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessInvoiceTagMapper.java
@@ -0,0 +1,33 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.UUID;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessInvoiceTag;
+
+public class BusinessInvoiceTagMapper implements ResultSetMapper<BusinessInvoiceTag> {
+    @Override
+    public BusinessInvoiceTag map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessInvoiceTag(UUID.fromString(r.getString(1)), r.getString(2));
+    }
+}

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
new file mode 100644
index 0000000..de9629a
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusBinder.java
@@ -0,0 +1,59 @@
+/*
+ * 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.lang.annotation.Annotation;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.sql.Types;
+
+import org.skife.jdbi.v2.SQLStatement;
+import org.skife.jdbi.v2.sqlobject.Binder;
+import org.skife.jdbi.v2.sqlobject.BinderFactory;
+import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
+
+import com.ning.billing.analytics.model.BusinessOverdueStatus;
+
+@BindingAnnotation(BusinessOverdueStatusBinder.BosBinderFactory.class)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.PARAMETER})
+public @interface BusinessOverdueStatusBinder {
+    public static class BosBinderFactory implements BinderFactory {
+        public Binder build(final Annotation annotation) {
+            return new Binder<BusinessOverdueStatusBinder, BusinessOverdueStatus>() {
+                public void bind(final SQLStatement q, final BusinessOverdueStatusBinder bind, final BusinessOverdueStatus overdueStatus) {
+                    q.bind("external_key", overdueStatus.getExternalKey());
+                    q.bind("status", overdueStatus.getStatus());
+
+                    if (overdueStatus.getStartDate() != null) {
+                        q.bind("start_date", overdueStatus.getStartDate());
+                    } else {
+                        q.bindNull("start_date", Types.BIGINT);
+                    }
+
+                    if (overdueStatus.getEndDate() != null) {
+                        q.bind("end_date", overdueStatus.getEndDate());
+                    } else {
+                        q.bindNull("end_date", Types.BIGINT);
+                    }
+                }
+            };
+        }
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusMapper.java
new file mode 100644
index 0000000..dcd6bf2
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessOverdueStatusMapper.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.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessOverdueStatus;
+
+public class BusinessOverdueStatusMapper implements ResultSetMapper<BusinessOverdueStatus> {
+    @Override
+    public BusinessOverdueStatus map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        final String externalKey = r.getString(1);
+        final String status = r.getString(2);
+        final DateTime startDate = new DateTime(r.getLong(3), DateTimeZone.UTC);
+        final DateTime endDate = new DateTime(r.getLong(4), DateTimeZone.UTC);
+
+        return new BusinessOverdueStatus(endDate, externalKey, startDate, status);
+    }
+}

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/BusinessSubscriptionTransitionBinder.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionBinder.java
index 2a04c8e..9e5b507 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionBinder.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionBinder.java
@@ -28,8 +28,8 @@ import org.skife.jdbi.v2.sqlobject.Binder;
 import org.skife.jdbi.v2.sqlobject.BinderFactory;
 import org.skife.jdbi.v2.sqlobject.BindingAnnotation;
 
-import com.ning.billing.analytics.BusinessSubscription;
-import com.ning.billing.analytics.BusinessSubscriptionTransition;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 
 @BindingAnnotation(BusinessSubscriptionTransitionBinder.BstBinderFactory.class)
 @Retention(RetentionPolicy.RUNTIME)
@@ -39,8 +39,8 @@ public @interface BusinessSubscriptionTransitionBinder {
         public Binder build(final Annotation annotation) {
             return new Binder<BusinessSubscriptionTransitionBinder, BusinessSubscriptionTransition>() {
                 public void bind(final SQLStatement q, final BusinessSubscriptionTransitionBinder bind, final BusinessSubscriptionTransition arg) {
-                    q.bind("event_id", arg.getId().toString());
-                    q.bind("event_key", arg.getKey());
+                    q.bind("total_ordering", arg.getTotalOrdering());
+                    q.bind("external_key", arg.getExternalKey());
                     q.bind("account_key", arg.getAccountKey());
                     q.bind("requested_timestamp", arg.getRequestedTimestamp().getMillis());
                     q.bind("event", arg.getEvent().toString());

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldMapper.java
new file mode 100644
index 0000000..68ef0e2
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionFieldMapper.java
@@ -0,0 +1,32 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionField;
+
+public class BusinessSubscriptionTransitionFieldMapper implements ResultSetMapper<BusinessSubscriptionTransitionField> {
+    @Override
+    public BusinessSubscriptionTransitionField map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessSubscriptionTransitionField(r.getString(1), r.getString(2), r.getString(3));
+    }
+}

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/BusinessSubscriptionTransitionMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionMapper.java
index ac38463..c5bdf96 100644
--- a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionMapper.java
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionMapper.java
@@ -26,9 +26,9 @@ import org.joda.time.DateTimeZone;
 import org.skife.jdbi.v2.StatementContext;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 
-import com.ning.billing.analytics.BusinessSubscription;
-import com.ning.billing.analytics.BusinessSubscriptionEvent;
-import com.ning.billing.analytics.BusinessSubscriptionTransition;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.catalog.api.ProductCategory;
 
 import static com.ning.billing.entitlement.api.user.Subscription.SubscriptionState;
@@ -83,7 +83,7 @@ public class BusinessSubscriptionTransitionMapper implements ResultSetMapper<Bus
         final BusinessSubscriptionEvent event = BusinessSubscriptionEvent.valueOf(r.getString(5));
 
         return new BusinessSubscriptionTransition(
-                UUID.fromString(r.getString(1)),
+                r.getLong(1),
                 r.getString(2),
                 r.getString(3),
                 new DateTime(r.getLong(4), DateTimeZone.UTC),

diff --git a/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagMapper.java b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagMapper.java
new file mode 100644
index 0000000..36f7631
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/dao/BusinessSubscriptionTransitionTagMapper.java
@@ -0,0 +1,32 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.skife.jdbi.v2.StatementContext;
+import org.skife.jdbi.v2.tweak.ResultSetMapper;
+
+import com.ning.billing.analytics.model.BusinessSubscriptionTransitionTag;
+
+public class BusinessSubscriptionTransitionTagMapper implements ResultSetMapper<BusinessSubscriptionTransitionTag> {
+    @Override
+    public BusinessSubscriptionTransitionTag map(final int index, final ResultSet r, final StatementContext ctx) throws SQLException {
+        return new BusinessSubscriptionTransitionTag(r.getString(1), r.getString(2));
+    }
+}

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/model/BusinessAccountField.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccountField.java
new file mode 100644
index 0000000..b990125
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccountField.java
@@ -0,0 +1,73 @@
+/*
+ * 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;
+
+public class BusinessAccountField extends BusinessField {
+    private final String accountKey;
+
+    public BusinessAccountField(final String accountKey, final String name, final String value) {
+        super(name, value);
+        this.accountKey = accountKey;
+    }
+
+    public String getAccountKey() {
+        return accountKey;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessAccountField");
+        sb.append("{accountKey='").append(accountKey).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append(", value='").append(value).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessAccountField that = (BusinessAccountField) o;
+
+        if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+        if (value != null ? !value.equals(that.value) : that.value != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = accountKey != null ? accountKey.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        result = 31 * result + (value != null ? value.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccountTag.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccountTag.java
new file mode 100644
index 0000000..3a8de99
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessAccountTag.java
@@ -0,0 +1,68 @@
+/*
+ * 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;
+
+public class BusinessAccountTag extends BusinessTag {
+    private final String accountKey;
+
+    public BusinessAccountTag(final String accountKey, final String name) {
+        super(name);
+        this.accountKey = accountKey;
+    }
+
+    public String getAccountKey() {
+        return accountKey;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessAccountTag");
+        sb.append("{accountKey='").append(accountKey).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessAccountTag that = (BusinessAccountTag) o;
+
+        if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = accountKey != null ? accountKey.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessField.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessField.java
new file mode 100644
index 0000000..68552d0
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessField.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+public abstract class BusinessField {
+    protected final String name;
+    protected final String value;
+
+    public BusinessField(final String name, final String value) {
+        this.name = name;
+        this.value = value;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getValue() {
+        return value;
+    }
+}

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
new file mode 100644
index 0000000..2c6e8bc
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoice.java
@@ -0,0 +1,219 @@
+/*
+ * 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 com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoice {
+    private final UUID invoiceId;
+    private final DateTime createdDate;
+
+    private DateTime updatedDate;
+    private String accountKey;
+    private DateTime invoiceDate;
+    private DateTime targetDate;
+    private Currency currency;
+    private BigDecimal balance;
+    private BigDecimal amountPaid;
+    private BigDecimal amountCharged;
+    private BigDecimal amountCredited;
+
+    public BusinessInvoice(final String accountKey, final BigDecimal amountCharged, final BigDecimal amountCredited,
+                           final BigDecimal amountPaid, final BigDecimal balance, final DateTime createdDate,
+                           final Currency currency, final DateTime invoiceDate, final UUID invoiceId,
+                           final DateTime targetDate, final DateTime updatedDate) {
+        this.accountKey = accountKey;
+        this.amountCharged = amountCharged;
+        this.amountCredited = amountCredited;
+        this.amountPaid = amountPaid;
+        this.balance = balance;
+        this.createdDate = createdDate;
+        this.currency = currency;
+        this.invoiceDate = invoiceDate;
+        this.invoiceId = invoiceId;
+        this.targetDate = targetDate;
+        this.updatedDate = updatedDate;
+    }
+
+    public DateTime getCreatedDate() {
+        return createdDate;
+    }
+
+    public UUID getInvoiceId() {
+        return invoiceId;
+    }
+
+    public String getAccountKey() {
+        return accountKey;
+    }
+
+    public void setAccountKey(final String accountKey) {
+        this.accountKey = accountKey;
+    }
+
+    public BigDecimal getAmountCharged() {
+        return amountCharged;
+    }
+
+    public void setAmountCharged(final BigDecimal amountCharged) {
+        this.amountCharged = amountCharged;
+    }
+
+    public BigDecimal getAmountCredited() {
+        return amountCredited;
+    }
+
+    public void setAmountCredited(final BigDecimal amountCredited) {
+        this.amountCredited = amountCredited;
+    }
+
+    public BigDecimal getAmountPaid() {
+        return amountPaid;
+    }
+
+    public void setAmountPaid(final BigDecimal amountPaid) {
+        this.amountPaid = amountPaid;
+    }
+
+    public BigDecimal getBalance() {
+        return balance;
+    }
+
+    public void setBalance(final BigDecimal balance) {
+        this.balance = balance;
+    }
+
+    public Currency getCurrency() {
+        return currency;
+    }
+
+    public void setCurrency(final Currency currency) {
+        this.currency = currency;
+    }
+
+    public DateTime getInvoiceDate() {
+        return invoiceDate;
+    }
+
+    public void setInvoiceDate(final DateTime invoiceDate) {
+        this.invoiceDate = invoiceDate;
+    }
+
+    public DateTime getTargetDate() {
+        return targetDate;
+    }
+
+    public void setTargetDate(final DateTime targetDate) {
+        this.targetDate = targetDate;
+    }
+
+    public DateTime getUpdatedDate() {
+        return updatedDate;
+    }
+
+    public void setUpdatedDate(final DateTime updatedDate) {
+        this.updatedDate = updatedDate;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessInvoice");
+        sb.append("{accountKey='").append(accountKey).append('\'');
+        sb.append(", invoiceId=").append(invoiceId);
+        sb.append(", createdDate=").append(createdDate);
+        sb.append(", updatedDate=").append(updatedDate);
+        sb.append(", invoiceDate=").append(invoiceDate);
+        sb.append(", targetDate=").append(targetDate);
+        sb.append(", currency=").append(currency);
+        sb.append(", balance=").append(balance);
+        sb.append(", amountPaid=").append(amountPaid);
+        sb.append(", amountCharged=").append(amountCharged);
+        sb.append(", amountCredited=").append(amountCredited);
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoice that = (BusinessInvoice) o;
+
+        if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
+            return false;
+        }
+        if (amountCharged != null ? !amountCharged.equals(that.amountCharged) : that.amountCharged != null) {
+            return false;
+        }
+        if (amountCredited != null ? !amountCredited.equals(that.amountCredited) : that.amountCredited != null) {
+            return false;
+        }
+        if (amountPaid != null ? !amountPaid.equals(that.amountPaid) : that.amountPaid != null) {
+            return false;
+        }
+        if (balance != null ? !balance.equals(that.balance) : that.balance != null) {
+            return false;
+        }
+        if (createdDate != null ? !createdDate.equals(that.createdDate) : that.createdDate != null) {
+            return false;
+        }
+        if (currency != that.currency) {
+            return false;
+        }
+        if (invoiceDate != null ? !invoiceDate.equals(that.invoiceDate) : that.invoiceDate != null) {
+            return false;
+        }
+        if (invoiceId != null ? !invoiceId.equals(that.invoiceId) : that.invoiceId != null) {
+            return false;
+        }
+        if (targetDate != null ? !targetDate.equals(that.targetDate) : that.targetDate != null) {
+            return false;
+        }
+        if (updatedDate != null ? !updatedDate.equals(that.updatedDate) : that.updatedDate != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = invoiceId != null ? invoiceId.hashCode() : 0;
+        result = 31 * result + (createdDate != null ? createdDate.hashCode() : 0);
+        result = 31 * result + (updatedDate != null ? updatedDate.hashCode() : 0);
+        result = 31 * result + (accountKey != null ? accountKey.hashCode() : 0);
+        result = 31 * result + (invoiceDate != null ? invoiceDate.hashCode() : 0);
+        result = 31 * result + (targetDate != null ? targetDate.hashCode() : 0);
+        result = 31 * result + (currency != null ? currency.hashCode() : 0);
+        result = 31 * result + (balance != null ? balance.hashCode() : 0);
+        result = 31 * result + (amountPaid != null ? amountPaid.hashCode() : 0);
+        result = 31 * result + (amountCharged != null ? amountCharged.hashCode() : 0);
+        result = 31 * result + (amountCredited != null ? amountCredited.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceField.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceField.java
new file mode 100644
index 0000000..cf69497
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceField.java
@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+public class BusinessInvoiceField extends BusinessField {
+    private final UUID invoiceId;
+
+    public BusinessInvoiceField(final UUID invoiceId, final String name, final String value) {
+        super(name, value);
+        this.invoiceId = invoiceId;
+    }
+
+    public UUID getInvoiceId() {
+        return invoiceId;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessInvoiceField");
+        sb.append("{invoiceId='").append(invoiceId).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append(", value='").append(value).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoiceField that = (BusinessInvoiceField) o;
+
+        if (invoiceId != null ? !invoiceId.equals(that.invoiceId) : that.invoiceId != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+        if (value != null ? !value.equals(that.value) : that.value != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = invoiceId != null ? invoiceId.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        result = 31 * result + (value != null ? value.hashCode() : 0);
+        return result;
+    }
+}

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
new file mode 100644
index 0000000..eb2793f
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceItem.java
@@ -0,0 +1,295 @@
+/*
+ * 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 com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoiceItem {
+    private final UUID itemId;
+    private final DateTime createdDate;
+
+    private DateTime updatedDate;
+    private UUID invoiceId;
+    private String itemType;
+    private String externalKey;
+    private String productName;
+    private String productType;
+    private String productCategory;
+    private String slug;
+    private String phase;
+    private String billingPeriod;
+    private DateTime startDate;
+    private DateTime endDate;
+    private BigDecimal amount;
+    private Currency currency;
+
+    public BusinessInvoiceItem(final BigDecimal amount, final String billingPeriod, final DateTime createdDate,
+                               final Currency currency, final DateTime endDate, final String externalKey,
+                               final UUID invoiceId, final UUID itemId, final String itemType, final String phase,
+                               final String productCategory, final String productName, final String productType,
+                               final String slug, final DateTime startDate, final DateTime updatedDate) {
+        this.amount = amount;
+        this.billingPeriod = billingPeriod;
+        this.createdDate = createdDate;
+        this.currency = currency;
+        this.endDate = endDate;
+        this.externalKey = externalKey;
+        this.invoiceId = invoiceId;
+        this.itemId = itemId;
+        this.itemType = itemType;
+        this.phase = phase;
+        this.productCategory = productCategory;
+        this.productName = productName;
+        this.productType = productType;
+        this.slug = slug;
+        this.startDate = startDate;
+        this.updatedDate = updatedDate;
+    }
+
+    public DateTime getCreatedDate() {
+        return createdDate;
+    }
+
+    public UUID getItemId() {
+        return itemId;
+    }
+
+    public BigDecimal getAmount() {
+        return amount;
+    }
+
+    public void setAmount(final BigDecimal amount) {
+        this.amount = amount;
+    }
+
+    public String getBillingPeriod() {
+        return billingPeriod;
+    }
+
+    public void setBillingPeriod(final String billingPeriod) {
+        this.billingPeriod = billingPeriod;
+    }
+
+    public Currency getCurrency() {
+        return currency;
+    }
+
+    public void setCurrency(final Currency currency) {
+        this.currency = currency;
+    }
+
+    public DateTime getEndDate() {
+        return endDate;
+    }
+
+    public void setEndDate(final DateTime endDate) {
+        this.endDate = endDate;
+    }
+
+    public String getExternalKey() {
+        return externalKey;
+    }
+
+    public void setExternalKey(final String externalKey) {
+        this.externalKey = externalKey;
+    }
+
+    public UUID getInvoiceId() {
+        return invoiceId;
+    }
+
+    public void setInvoiceId(final UUID invoiceId) {
+        this.invoiceId = invoiceId;
+    }
+
+    public String getItemType() {
+        return itemType;
+    }
+
+    public void setItemType(final String itemType) {
+        this.itemType = itemType;
+    }
+
+    public String getPhase() {
+        return phase;
+    }
+
+    public void setPhase(final String phase) {
+        this.phase = phase;
+    }
+
+    public String getProductCategory() {
+        return productCategory;
+    }
+
+    public void setProductCategory(final String productCategory) {
+        this.productCategory = productCategory;
+    }
+
+    public String getProductName() {
+        return productName;
+    }
+
+    public void setProductName(final String productName) {
+        this.productName = productName;
+    }
+
+    public String getProductType() {
+        return productType;
+    }
+
+    public void setProductType(final String productType) {
+        this.productType = productType;
+    }
+
+    public String getSlug() {
+        return slug;
+    }
+
+    public void setSlug(final String slug) {
+        this.slug = slug;
+    }
+
+    public DateTime getStartDate() {
+        return startDate;
+    }
+
+    public void setStartDate(final DateTime startDate) {
+        this.startDate = startDate;
+    }
+
+    public DateTime getUpdatedDate() {
+        return updatedDate;
+    }
+
+    public void setUpdatedDate(final DateTime updatedDate) {
+        this.updatedDate = updatedDate;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessInvoiceItem");
+        sb.append("{amount=").append(amount);
+        sb.append(", itemId=").append(itemId);
+        sb.append(", createdDate=").append(createdDate);
+        sb.append(", updatedDate=").append(updatedDate);
+        sb.append(", invoiceId=").append(invoiceId);
+        sb.append(", itemType='").append(itemType).append('\'');
+        sb.append(", externalKey='").append(externalKey).append('\'');
+        sb.append(", productName='").append(productName).append('\'');
+        sb.append(", productType='").append(productType).append('\'');
+        sb.append(", productCategory='").append(productCategory).append('\'');
+        sb.append(", slug='").append(slug).append('\'');
+        sb.append(", phase='").append(phase).append('\'');
+        sb.append(", billingPeriod='").append(billingPeriod).append('\'');
+        sb.append(", startDate=").append(startDate);
+        sb.append(", endDate=").append(endDate);
+        sb.append(", currency=").append(currency);
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoiceItem that = (BusinessInvoiceItem) o;
+
+        if (amount != null ? !amount.equals(that.amount) : that.amount != null) {
+            return false;
+        }
+        if (billingPeriod != null ? !billingPeriod.equals(that.billingPeriod) : that.billingPeriod != null) {
+            return false;
+        }
+        if (createdDate != null ? !createdDate.equals(that.createdDate) : that.createdDate != null) {
+            return false;
+        }
+        if (currency != that.currency) {
+            return false;
+        }
+        if (endDate != null ? !endDate.equals(that.endDate) : that.endDate != null) {
+            return false;
+        }
+        if (externalKey != null ? !externalKey.equals(that.externalKey) : that.externalKey != null) {
+            return false;
+        }
+        if (invoiceId != null ? !invoiceId.equals(that.invoiceId) : that.invoiceId != null) {
+            return false;
+        }
+        if (itemId != null ? !itemId.equals(that.itemId) : that.itemId != null) {
+            return false;
+        }
+        if (itemType != null ? !itemType.equals(that.itemType) : that.itemType != null) {
+            return false;
+        }
+        if (phase != null ? !phase.equals(that.phase) : that.phase != null) {
+            return false;
+        }
+        if (productCategory != null ? !productCategory.equals(that.productCategory) : that.productCategory != null) {
+            return false;
+        }
+        if (productName != null ? !productName.equals(that.productName) : that.productName != null) {
+            return false;
+        }
+        if (productType != null ? !productType.equals(that.productType) : that.productType != null) {
+            return false;
+        }
+        if (slug != null ? !slug.equals(that.slug) : that.slug != null) {
+            return false;
+        }
+        if (startDate != null ? !startDate.equals(that.startDate) : that.startDate != null) {
+            return false;
+        }
+        if (updatedDate != null ? !updatedDate.equals(that.updatedDate) : that.updatedDate != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = itemId != null ? itemId.hashCode() : 0;
+        result = 31 * result + (createdDate != null ? createdDate.hashCode() : 0);
+        result = 31 * result + (updatedDate != null ? updatedDate.hashCode() : 0);
+        result = 31 * result + (invoiceId != null ? invoiceId.hashCode() : 0);
+        result = 31 * result + (itemType != null ? itemType.hashCode() : 0);
+        result = 31 * result + (externalKey != null ? externalKey.hashCode() : 0);
+        result = 31 * result + (productName != null ? productName.hashCode() : 0);
+        result = 31 * result + (productType != null ? productType.hashCode() : 0);
+        result = 31 * result + (productCategory != null ? productCategory.hashCode() : 0);
+        result = 31 * result + (slug != null ? slug.hashCode() : 0);
+        result = 31 * result + (phase != null ? phase.hashCode() : 0);
+        result = 31 * result + (billingPeriod != null ? billingPeriod.hashCode() : 0);
+        result = 31 * result + (startDate != null ? startDate.hashCode() : 0);
+        result = 31 * result + (endDate != null ? endDate.hashCode() : 0);
+        result = 31 * result + (amount != null ? amount.hashCode() : 0);
+        result = 31 * result + (currency != null ? currency.hashCode() : 0);
+        return result;
+    }
+}

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
new file mode 100644
index 0000000..a0ea526
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePayment.java
@@ -0,0 +1,307 @@
+/*
+ * 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 com.ning.billing.catalog.api.Currency;
+
+public class BusinessInvoicePayment {
+    private final UUID paymentId;
+    private final DateTime createdDate;
+    private final UUID attemptId;
+
+    private DateTime updatedDate;
+    private String accountKey;
+    private UUID invoiceId;
+    private DateTime effectiveDate;
+    private BigDecimal amount;
+    private Currency currency;
+    private String paymentError;
+    private String processingStatus;
+    private BigDecimal requestedAmount;
+    private String pluginName;
+    private String paymentType;
+    private String paymentMethod;
+    private String cardType;
+    private String cardCountry;
+
+    public BusinessInvoicePayment(final String accountKey, final BigDecimal amount, final UUID attemptId,
+                                  final String cardCountry, final String cardType, final DateTime createdDate,
+                                  final Currency currency, final DateTime effectiveDate, final UUID invoiceId,
+                                  final String paymentError, final UUID paymentId, final String paymentMethod,
+                                  final String paymentType, final String pluginName, final String processingStatus,
+                                  final BigDecimal requestedAmount, final DateTime updatedDate) {
+        this.accountKey = accountKey;
+        this.amount = amount;
+        this.attemptId = attemptId;
+        this.cardCountry = cardCountry;
+        this.cardType = cardType;
+        this.createdDate = createdDate;
+        this.currency = currency;
+        this.effectiveDate = effectiveDate;
+        this.invoiceId = invoiceId;
+        this.paymentError = paymentError;
+        this.paymentId = paymentId;
+        this.paymentMethod = paymentMethod;
+        this.paymentType = paymentType;
+        this.pluginName = pluginName;
+        this.processingStatus = processingStatus;
+        this.requestedAmount = requestedAmount;
+        this.updatedDate = updatedDate;
+    }
+
+    public UUID getAttemptId() {
+        return attemptId;
+    }
+
+    public DateTime getCreatedDate() {
+        return createdDate;
+    }
+
+    public UUID getPaymentId() {
+        return paymentId;
+    }
+
+    public String getAccountKey() {
+        return accountKey;
+    }
+
+    public void setAccountKey(final String accountKey) {
+        this.accountKey = accountKey;
+    }
+
+    public BigDecimal getAmount() {
+        return amount;
+    }
+
+    public void setAmount(final BigDecimal amount) {
+        this.amount = amount;
+    }
+
+    public String getCardCountry() {
+        return cardCountry;
+    }
+
+    public void setCardCountry(final String cardCountry) {
+        this.cardCountry = cardCountry;
+    }
+
+    public String getCardType() {
+        return cardType;
+    }
+
+    public void setCardType(final String cardType) {
+        this.cardType = cardType;
+    }
+
+    public Currency getCurrency() {
+        return currency;
+    }
+
+    public void setCurrency(final Currency currency) {
+        this.currency = currency;
+    }
+
+    public DateTime getEffectiveDate() {
+        return effectiveDate;
+    }
+
+    public void setEffectiveDate(final DateTime effectiveDate) {
+        this.effectiveDate = effectiveDate;
+    }
+
+    public UUID getInvoiceId() {
+        return invoiceId;
+    }
+
+    public void setInvoiceId(final UUID invoiceId) {
+        this.invoiceId = invoiceId;
+    }
+
+    public String getPaymentError() {
+        return paymentError;
+    }
+
+    public void setPaymentError(final String paymentError) {
+        this.paymentError = paymentError;
+    }
+
+    public String getPaymentMethod() {
+        return paymentMethod;
+    }
+
+    public void setPaymentMethod(final String paymentMethod) {
+        this.paymentMethod = paymentMethod;
+    }
+
+    public String getPaymentType() {
+        return paymentType;
+    }
+
+    public void setPaymentType(final String paymentType) {
+        this.paymentType = paymentType;
+    }
+
+    public String getPluginName() {
+        return pluginName;
+    }
+
+    public void setPluginName(final String pluginName) {
+        this.pluginName = pluginName;
+    }
+
+    public String getProcessingStatus() {
+        return processingStatus;
+    }
+
+    public void setProcessingStatus(final String processingStatus) {
+        this.processingStatus = processingStatus;
+    }
+
+    public BigDecimal getRequestedAmount() {
+        return requestedAmount;
+    }
+
+    public void setRequestedAmount(final BigDecimal requestedAmount) {
+        this.requestedAmount = requestedAmount;
+    }
+
+    public DateTime getUpdatedDate() {
+        return updatedDate;
+    }
+
+    public void setUpdatedDate(final DateTime updatedDate) {
+        this.updatedDate = updatedDate;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessInvoicePayment");
+        sb.append("{accountKey='").append(accountKey).append('\'');
+        sb.append(", paymentId=").append(paymentId);
+        sb.append(", createdDate=").append(createdDate);
+        sb.append(", attemptId=").append(attemptId);
+        sb.append(", updatedDate=").append(updatedDate);
+        sb.append(", invoiceId=").append(invoiceId);
+        sb.append(", effectiveDate=").append(effectiveDate);
+        sb.append(", amount=").append(amount);
+        sb.append(", currency=").append(currency);
+        sb.append(", paymentError='").append(paymentError).append('\'');
+        sb.append(", processingStatus='").append(processingStatus).append('\'');
+        sb.append(", requestedAmount=").append(requestedAmount);
+        sb.append(", pluginName='").append(pluginName).append('\'');
+        sb.append(", paymentType='").append(paymentType).append('\'');
+        sb.append(", paymentMethod='").append(paymentMethod).append('\'');
+        sb.append(", cardType='").append(cardType).append('\'');
+        sb.append(", cardCountry='").append(cardCountry).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoicePayment that = (BusinessInvoicePayment) o;
+
+        if (accountKey != null ? !accountKey.equals(that.accountKey) : that.accountKey != null) {
+            return false;
+        }
+        if (amount != null ? !amount.equals(that.amount) : that.amount != null) {
+            return false;
+        }
+        if (attemptId != null ? !attemptId.equals(that.attemptId) : that.attemptId != null) {
+            return false;
+        }
+        if (cardCountry != null ? !cardCountry.equals(that.cardCountry) : that.cardCountry != null) {
+            return false;
+        }
+        if (cardType != null ? !cardType.equals(that.cardType) : that.cardType != null) {
+            return false;
+        }
+        if (createdDate != null ? !createdDate.equals(that.createdDate) : that.createdDate != null) {
+            return false;
+        }
+        if (currency != that.currency) {
+            return false;
+        }
+        if (effectiveDate != null ? !effectiveDate.equals(that.effectiveDate) : that.effectiveDate != null) {
+            return false;
+        }
+        if (invoiceId != null ? !invoiceId.equals(that.invoiceId) : that.invoiceId != null) {
+            return false;
+        }
+        if (paymentError != null ? !paymentError.equals(that.paymentError) : that.paymentError != null) {
+            return false;
+        }
+        if (paymentId != null ? !paymentId.equals(that.paymentId) : that.paymentId != null) {
+            return false;
+        }
+        if (paymentMethod != null ? !paymentMethod.equals(that.paymentMethod) : that.paymentMethod != null) {
+            return false;
+        }
+        if (paymentType != null ? !paymentType.equals(that.paymentType) : that.paymentType != null) {
+            return false;
+        }
+        if (pluginName != null ? !pluginName.equals(that.pluginName) : that.pluginName != null) {
+            return false;
+        }
+        if (processingStatus != null ? !processingStatus.equals(that.processingStatus) : that.processingStatus != null) {
+            return false;
+        }
+        if (requestedAmount != null ? !requestedAmount.equals(that.requestedAmount) : that.requestedAmount != null) {
+            return false;
+        }
+        if (updatedDate != null ? !updatedDate.equals(that.updatedDate) : that.updatedDate != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = paymentId != null ? paymentId.hashCode() : 0;
+        result = 31 * result + (createdDate != null ? createdDate.hashCode() : 0);
+        result = 31 * result + (attemptId != null ? attemptId.hashCode() : 0);
+        result = 31 * result + (updatedDate != null ? updatedDate.hashCode() : 0);
+        result = 31 * result + (accountKey != null ? accountKey.hashCode() : 0);
+        result = 31 * result + (invoiceId != null ? invoiceId.hashCode() : 0);
+        result = 31 * result + (effectiveDate != null ? effectiveDate.hashCode() : 0);
+        result = 31 * result + (amount != null ? amount.hashCode() : 0);
+        result = 31 * result + (currency != null ? currency.hashCode() : 0);
+        result = 31 * result + (paymentError != null ? paymentError.hashCode() : 0);
+        result = 31 * result + (processingStatus != null ? processingStatus.hashCode() : 0);
+        result = 31 * result + (requestedAmount != null ? requestedAmount.hashCode() : 0);
+        result = 31 * result + (pluginName != null ? pluginName.hashCode() : 0);
+        result = 31 * result + (paymentType != null ? paymentType.hashCode() : 0);
+        result = 31 * result + (paymentMethod != null ? paymentMethod.hashCode() : 0);
+        result = 31 * result + (cardType != null ? cardType.hashCode() : 0);
+        result = 31 * result + (cardCountry != null ? cardCountry.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentField.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentField.java
new file mode 100644
index 0000000..bcfc85b
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentField.java
@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+public class BusinessInvoicePaymentField extends BusinessField {
+    private final UUID paymentId;
+
+    public BusinessInvoicePaymentField(final UUID paymentId, final String name, final String value) {
+        super(name, value);
+        this.paymentId = paymentId;
+    }
+
+    public UUID getPaymentId() {
+        return paymentId;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessPaymentField");
+        sb.append("{paymentId='").append(paymentId).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append(", value='").append(value).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoicePaymentField that = (BusinessInvoicePaymentField) o;
+
+        if (paymentId != null ? !paymentId.equals(that.paymentId) : that.paymentId != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+        if (value != null ? !value.equals(that.value) : that.value != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = paymentId != null ? paymentId.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        result = 31 * result + (value != null ? value.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentTag.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentTag.java
new file mode 100644
index 0000000..fd48b1c
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoicePaymentTag.java
@@ -0,0 +1,70 @@
+/*
+ * 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;
+
+public class BusinessInvoicePaymentTag extends BusinessTag {
+    private final UUID paymentId;
+
+    public BusinessInvoicePaymentTag(final UUID paymentId, final String name) {
+        super(name);
+        this.paymentId = paymentId;
+    }
+
+    public UUID getPaymentId() {
+        return paymentId;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessPaymentTag");
+        sb.append("{paymentId='").append(paymentId).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoicePaymentTag that = (BusinessInvoicePaymentTag) o;
+
+        if (paymentId != null ? !paymentId.equals(that.paymentId) : that.paymentId != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = paymentId != null ? paymentId.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceTag.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceTag.java
new file mode 100644
index 0000000..e9efe5a
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessInvoiceTag.java
@@ -0,0 +1,70 @@
+/*
+ * 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;
+
+public class BusinessInvoiceTag extends BusinessTag {
+    private final UUID invoiceId;
+
+    public BusinessInvoiceTag(final UUID invoiceId, final String name) {
+        super(name);
+        this.invoiceId = invoiceId;
+    }
+
+    public UUID getInvoiceId() {
+        return invoiceId;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessInvoiceTag");
+        sb.append("{paymentId='").append(invoiceId).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessInvoiceTag that = (BusinessInvoiceTag) o;
+
+        if (invoiceId != null ? !invoiceId.equals(that.invoiceId) : that.invoiceId != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = invoiceId != null ? invoiceId.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessOverdueStatus.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessOverdueStatus.java
new file mode 100644
index 0000000..dd8a310
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessOverdueStatus.java
@@ -0,0 +1,97 @@
+/*
+ * 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 org.joda.time.DateTime;
+
+public class BusinessOverdueStatus {
+    private final String externalKey;
+    private final String status;
+    private final DateTime startDate;
+    private final DateTime endDate;
+
+    public BusinessOverdueStatus(final DateTime endDate, final String externalKey, final DateTime startDate, final String status) {
+        this.endDate = endDate;
+        this.externalKey = externalKey;
+        this.startDate = startDate;
+        this.status = status;
+    }
+
+    public DateTime getEndDate() {
+        return endDate;
+    }
+
+    public String getExternalKey() {
+        return externalKey;
+    }
+
+    public DateTime getStartDate() {
+        return startDate;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessOverdueStatus");
+        sb.append("{endDate=").append(endDate);
+        sb.append(", externalKey='").append(externalKey).append('\'');
+        sb.append(", status='").append(status).append('\'');
+        sb.append(", startDate=").append(startDate);
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessOverdueStatus that = (BusinessOverdueStatus) o;
+
+        if (endDate != null ? !endDate.equals(that.endDate) : that.endDate != null) {
+            return false;
+        }
+        if (externalKey != null ? !externalKey.equals(that.externalKey) : that.externalKey != null) {
+            return false;
+        }
+        if (startDate != null ? !startDate.equals(that.startDate) : that.startDate != null) {
+            return false;
+        }
+        if (status != null ? !status.equals(that.status) : that.status != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = externalKey != null ? externalKey.hashCode() : 0;
+        result = 31 * result + (status != null ? status.hashCode() : 0);
+        result = 31 * result + (startDate != null ? startDate.hashCode() : 0);
+        result = 31 * result + (endDate != null ? endDate.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionField.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionField.java
new file mode 100644
index 0000000..bab1259
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionField.java
@@ -0,0 +1,73 @@
+/*
+ * 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;
+
+public class BusinessSubscriptionTransitionField extends BusinessField {
+    private final String externalKey;
+
+    public BusinessSubscriptionTransitionField(final String externalKey, final String name, final String value) {
+        super(name, value);
+        this.externalKey = externalKey;
+    }
+
+    public String getExternalKey() {
+        return externalKey;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessSubscriptionTransitionField");
+        sb.append("{externalKey='").append(externalKey).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append(", value='").append(value).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessSubscriptionTransitionField that = (BusinessSubscriptionTransitionField) o;
+
+        if (externalKey != null ? !externalKey.equals(that.externalKey) : that.externalKey != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+        if (value != null ? !value.equals(that.value) : that.value != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = externalKey != null ? externalKey.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        result = 31 * result + (value != null ? value.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionTag.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionTag.java
new file mode 100644
index 0000000..522194a
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessSubscriptionTransitionTag.java
@@ -0,0 +1,68 @@
+/*
+ * 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;
+
+public class BusinessSubscriptionTransitionTag extends BusinessTag {
+    private final String externalKey;
+
+    public BusinessSubscriptionTransitionTag(final String externalKey, final String name) {
+        super(name);
+        this.externalKey = externalKey;
+    }
+
+    public String getExternalKey() {
+        return externalKey;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("BusinessSubscriptionTransitionTag");
+        sb.append("{externalKey='").append(externalKey).append('\'');
+        sb.append(", name='").append(name).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        final BusinessSubscriptionTransitionTag that = (BusinessSubscriptionTransitionTag) o;
+
+        if (externalKey != null ? !externalKey.equals(that.externalKey) : that.externalKey != null) {
+            return false;
+        }
+        if (name != null ? !name.equals(that.name) : that.name != null) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = externalKey != null ? externalKey.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        return result;
+    }
+}

diff --git a/analytics/src/main/java/com/ning/billing/analytics/model/BusinessTag.java b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessTag.java
new file mode 100644
index 0000000..80343e1
--- /dev/null
+++ b/analytics/src/main/java/com/ning/billing/analytics/model/BusinessTag.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+public abstract class BusinessTag {
+    protected final String name;
+
+    public BusinessTag(final String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+}

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 2233e1e..bb564e6 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
@@ -23,16 +23,16 @@ import com.ning.billing.analytics.BusinessAccountRecorder;
 import com.ning.billing.analytics.BusinessSubscriptionTransitionRecorder;
 import com.ning.billing.analytics.api.AnalyticsService;
 import com.ning.billing.analytics.api.DefaultAnalyticsService;
-import com.ning.billing.analytics.dao.BusinessAccountDao;
-import com.ning.billing.analytics.dao.BusinessAccountDaoProvider;
-import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionDao;
-import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionDaoProvider;
+import com.ning.billing.analytics.dao.BusinessAccountSqlDao;
+import com.ning.billing.analytics.dao.BusinessAccountSqlDaoProvider;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDaoProvider;
 
 public class AnalyticsModule extends AbstractModule {
     @Override
     protected void configure() {
-        bind(BusinessSubscriptionTransitionDao.class).toProvider(BusinessSubscriptionTransitionDaoProvider.class).asEagerSingleton();
-        bind(BusinessAccountDao.class).toProvider(BusinessAccountDaoProvider.class).asEagerSingleton();
+        bind(BusinessSubscriptionTransitionSqlDao.class).toProvider(BusinessSubscriptionTransitionSqlDaoProvider.class).asEagerSingleton();
+        bind(BusinessAccountSqlDao.class).toProvider(BusinessAccountSqlDaoProvider.class).asEagerSingleton();
 
         bind(BusinessSubscriptionTransitionRecorder.class).asEagerSingleton();
         bind(BusinessAccountRecorder.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..2924db3
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceFieldSqlDao.sql.stg
@@ -0,0 +1,30 @@
+group BusinessInvoiceField;
+
+getFieldsForInvoice(invoice_id) ::=<<
+select
+  invoice_id
+, name
+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
new file mode 100644
index 0000000..44862ea
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceItemSqlDao.sql.stg
@@ -0,0 +1,137 @@
+group BusinessInvoiceItem;
+
+getInvoiceItem(item_id) ::= <<
+select
+  item_id
+, created_date
+, updated_date
+, invoice_id
+, item_type
+, external_key
+, product_name
+, product_type
+, product_category
+, slug
+, phase
+, billing_period
+, start_date
+, end_date
+, amount
+, currency
+from bii
+where item_id = :item_id
+limit 1
+;
+>>
+
+getInvoiceItemsForInvoice(invoice_id) ::= <<
+select
+  item_id
+, created_date
+, updated_date
+, invoice_id
+, item_type
+, external_key
+, product_name
+, product_type
+, product_category
+, slug
+, phase
+, billing_period
+, start_date
+, end_date
+, amount
+, currency
+from bii
+where invoice_id = :invoice_id
+;
+>>
+
+getInvoiceItemsForBundle(external_key) ::= <<
+select
+  item_id
+, created_date
+, updated_date
+, invoice_id
+, item_type
+, external_key
+, product_name
+, product_type
+, product_category
+, slug
+, phase
+, billing_period
+, start_date
+, end_date
+, amount
+, currency
+from bii
+where external_key = :external_key
+;
+>>
+
+createInvoiceItem() ::= <<
+insert into bii (
+  item_id
+, created_date
+, updated_date
+, invoice_id
+, item_type
+, external_key
+, product_name
+, product_type
+, product_category
+, slug
+, phase
+, billing_period
+, start_date
+, end_date
+, amount
+, currency
+) values (
+  :item_id
+, :created_date
+, :updated_date
+, :invoice_id
+, :item_type
+, :external_key
+, :product_name
+, :product_type
+, :product_category
+, :slug
+, :phase
+, :billing_period
+, :start_date
+, :end_date
+, :amount
+, :currency
+);
+>>
+
+updateInvoiceItem() ::= <<
+update bii set
+  updated_date = :updated_date
+, invoice_id = :invoice_id
+, item_type = :item_type
+, external_key = :external_key
+, product_name = :product_name
+, product_type = :product_type
+, product_category = :product_category
+, slug = :slug
+, phase = :phase
+, billing_period = :billing_period
+, start_date = :start_date
+, end_date = :end_date
+, amount = :amount
+, currency = :currency
+where item_id = :item_id
+;
+>>
+
+deleteInvoiceItem(item_id) ::= <<
+delete from bii where item_id = :item_id;
+>>
+
+test() ::= <<
+select 1 from bii;
+>>
\ No newline at end of file

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
new file mode 100644
index 0000000..e4b800d
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoicePaymentSqlDao.sql.stg
@@ -0,0 +1,142 @@
+group BusinessInvoicePayment;
+
+getInvoicePaymentForPaymentAttempt(attempt_id) ::= <<
+select
+  payment_id
+, created_date
+, updated_date
+, attempt_id
+, account_key
+, invoice_id
+, effective_date
+, amount
+, currency
+, payment_error
+, processing_status
+, requested_amount
+, plugin_name
+, payment_type
+, payment_method
+, card_type
+, card_country
+from bip
+where attempt_id = :attempt_id
+limit 1
+;
+>>
+
+getInvoicePaymentsForPayment(payment_id) ::= <<
+select
+  payment_id
+, created_date
+, updated_date
+, attempt_id
+, account_key
+, invoice_id
+, effective_date
+, amount
+, currency
+, payment_error
+, processing_status
+, requested_amount
+, plugin_name
+, payment_type
+, payment_method
+, card_type
+, card_country
+from bip
+where payment_id = :payment_id
+;
+>>
+
+getInvoicePaymentsForAccount(account_key) ::= <<
+select
+  payment_id
+, created_date
+, updated_date
+, attempt_id
+, account_key
+, invoice_id
+, effective_date
+, amount
+, currency
+, payment_error
+, processing_status
+, requested_amount
+, plugin_name
+, payment_type
+, payment_method
+, card_type
+, card_country
+from bip
+where account_key = :account_key
+;
+>>
+
+createInvoicePayment() ::= <<
+insert into bip (
+  payment_id
+, created_date
+, updated_date
+, attempt_id
+, account_key
+, invoice_id
+, effective_date
+, amount
+, currency
+, payment_error
+, processing_status
+, requested_amount
+, plugin_name
+, payment_type
+, payment_method
+, card_type
+, card_country
+) values (
+  :payment_id
+, :created_date
+, :updated_date
+, :attempt_id
+, :account_key
+, :invoice_id
+, :effective_date
+, :amount
+, :currency
+, :payment_error
+, :processing_status
+, :requested_amount
+, :plugin_name
+, :payment_type
+, :payment_method
+, :card_type
+, :card_country
+);
+>>
+
+updateInvoicePaymentForPaymentAttempt() ::= <<
+update bip set
+  updated_date = :updated_date
+, account_key = :account_key
+, invoice_id = :invoice_id
+, effective_date = :effective_date
+, amount = :amount
+, currency = :currency
+, payment_error = :payment_error
+, processing_status = :processing_status
+, requested_amount = :requested_amount
+, plugin_name = :plugin_name
+, payment_type = :payment_type
+, payment_method = :payment_method
+, card_type = :card_type
+, card_country = :card_country
+where attempt_id = :attempt_id
+;
+>>
+
+deleteInvoicePaymentForPaymentAttempt(attempt_id) ::= <<
+delete from bip where attempt_id = :attempt_id
+>>
+
+test() ::= <<
+select 1 from bip;
+>>
\ No newline at end of file

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
new file mode 100644
index 0000000..7d72ef0
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessInvoiceSqlDao.sql.stg
@@ -0,0 +1,89 @@
+group BusinessInvoice;
+
+getInvoice(invoice_id) ::= <<
+select
+  invoice_id
+, created_date
+, updated_date
+, account_key
+, invoice_date
+, target_date
+, currency
+, balance
+, amount_paid
+, amount_charged
+, amount_credited
+from bin
+where invoice_id = :invoice_id
+limit 1
+;
+>>
+
+getInvoicesForAccount(account_key) ::= <<
+select
+  invoice_id
+, created_date
+, updated_date
+, account_key
+, invoice_date
+, target_date
+, currency
+, balance
+, amount_paid
+, amount_charged
+, amount_credited
+from bin
+where account_key = :account_key
+;
+>>
+
+createInvoice() ::= <<
+insert into bin (
+  invoice_id
+, created_date
+, updated_date
+, account_key
+, invoice_date
+, target_date
+, currency
+, balance
+, amount_paid
+, amount_charged
+, amount_credited
+) values (
+  :invoice_id
+, :created_date
+, :updated_date
+, :account_key
+, :invoice_date
+, :target_date
+, :currency
+, :balance
+, :amount_paid
+, :amount_charged
+, :amount_credited
+);
+>>
+
+updateInvoice() ::= <<
+update bin set
+  updated_date = :updated_date
+, account_key = :account_key
+, invoice_date = :invoice_date
+, target_date = :target_date
+, currency = :currency
+, balance = :balance
+, amount_paid = :amount_paid
+, amount_charged = :amount_charged
+, amount_credited = :amount_credited
+where invoice_id = :invoice_id
+;
+>>
+
+deleteInvoice(invoice_id) ::= <<
+delete from bin where invoice_id = :invoice_id;
+>>
+
+test() ::= <<
+select 1 from bin;
+>>
\ No newline at end of file

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
new file mode 100644
index 0000000..7153cc2
--- /dev/null
+++ b/analytics/src/main/resources/com/ning/billing/analytics/dao/BusinessOverdueStatusSqlDao.sql.stg
@@ -0,0 +1,30 @@
+group BusinessOverdueStatus;
+
+getOverdueStatusesForBundle(external_key) ::= <<
+select
+  external_key
+, status
+, start_date
+, end_date
+from bos
+where external_key = :external_key
+;
+>>
+
+createOverdueStatus() ::= <<
+insert into bos (
+  external_key
+, status
+, start_date
+, end_date
+) values (
+  :external_key
+, :status
+, :start_date
+, :end_date
+);
+>>
+
+test() ::= <<
+select 1 from bos;
+>>
\ No newline at end of file

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 7240236..43aaad8 100644
--- a/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql
+++ b/analytics/src/main/resources/com/ning/billing/analytics/ddl.sql
@@ -1,8 +1,8 @@
 drop table if exists bst;
 create table bst (
-  event_id char(36) not null
-, event_key varchar(50) not null
-, account_key varchar(50) not null
+  total_ordering bigint default 0
+, external_key varchar(50) not null comment 'Bundle external key'
+, account_key varchar(50) not null comment 'Account external key'
 , requested_timestamp bigint not null
 , event varchar(50) not null
 , prev_product_name varchar(32) default null
@@ -33,22 +33,143 @@ create table bst (
 , next_state varchar(32) default null
 , next_subscription_id varchar(100) default null
 , next_bundle_id varchar(100) default null
-, primary key(event_id)
-) engine=innodb;
-create index bst_key_index on bst (event_key, requested_timestamp asc);
+, primary key(total_ordering)
+) 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;
 create table bac (
   account_key varchar(50) not null
+, name varchar(100) not null
 , created_date bigint not null
 , updated_date bigint not null
 , balance numeric(10, 4) default 0
-, tags varchar(500) default null
 , last_invoice_date bigint default null
 , total_invoice_balance numeric(10, 4) default 0
 , last_payment_status varchar(100) default null
 , payment_method varchar(100) default null
 , credit_card_type varchar(32) default null
 , billing_address_country varchar(100) default null
-) engine=innodb;
-create unique index bac_key_index on bac (account_key);
\ No newline at end of file
+, currency char(3) default null
+) engine=innodb comment 'Business ACcounts, keep a record of all accounts';
+create unique index bac_key_index on bac (account_key);
+
+drop table if exists bin;
+create table bin (
+  invoice_id char(36) not null
+, created_date bigint not null
+, updated_date bigint not null
+, account_key varchar(50) not null
+, invoice_date bigint not null
+, target_date bigint not null
+, currency char(3) not null
+, balance numeric(10, 4) default 0 comment 'amount_charged - amount_paid - amount_credited'
+, amount_paid numeric(10, 4) default 0 comment 'Sums of the successful payments made for this invoice minus the refunds associated with this invoice'
+, amount_charged numeric(10, 4) default 0 comment 'Sums of the invoice items amount'
+, amount_credited numeric(10, 4) default 0 comment 'Sums of the credit items'
+) engine=innodb comment 'Business INvoices, keep a record of generated invoices';
+create unique index bin_key_index on bin (invoice_id);
+
+drop table if exists bii;
+create table bii (
+  item_id char(36) not null
+, created_date bigint not null
+, updated_date bigint not null
+, invoice_id char(36) not null
+, item_type char(20) not null comment 'e.g. FIXED or RECURRING'
+, external_key varchar(50) not null comment 'Bundle external key'
+, product_name varchar(32) default null
+, product_type varchar(32) default null
+, product_category varchar(32) default null
+, slug varchar(50) default null comment 'foo'
+, phase varchar(32) default null
+, billing_period varchar(32) default null
+, start_date bigint default null
+, end_date bigint default null
+, amount numeric(10, 4) default 0
+, currency char(3) default null
+) engine=innodb comment 'Business Invoice Items, keep a record of all invoice items';
+create unique index bii_key_index on bii (item_id);
+
+drop table if exists bip;
+create table bip (
+  payment_id char(36) not null
+, created_date bigint not null
+, updated_date bigint not null
+, attempt_id char(36) not null
+, account_key varchar(50) not null comment 'Account external key'
+, invoice_id char(36) not null
+, effective_date bigint default null
+, amount numeric(10, 4) default 0
+, currency char(3) default null
+, payment_error varchar(256) default null
+, processing_status varchar(50) default null
+, requested_amount numeric(10, 4) default 0
+, plugin_name varchar(20) default null
+, payment_type varchar(20) default null
+, payment_method varchar(20) default null
+, card_type varchar(20) default null
+, card_country varchar(20) default null
+) engine=innodb comment 'Business Invoice Payments, track all payment attempts';
+create unique index bip_key_index on bip (attempt_id);
+
+drop table if exists bos;
+create table bos (
+  external_key varchar(50) not null comment 'Bundle external key'
+, status varchar(50) not null
+, start_date bigint default null
+, end_date bigint default null
+) engine=innodb comment 'Business Overdue Status, historical bundles overdue status';
+create unique index bos_key_index on bos (external_key, status);
+
+drop table if exists bac_tags;
+create table bac_tags (
+  account_key varchar(50) not null comment 'Account external key'
+, name varchar(20) not null
+) engine=innodb comment 'Tags associated to accounts';
+
+drop table if exists bac_fields;
+create table bac_fields (
+  account_key varchar(50) not null comment 'Account external key'
+, name varchar(30) not null
+, value varchar(255) default null
+) engine=innodb comment 'Custom fields associated to accounts';
+
+drop table if exists bst_tags;
+create table bst_tags (
+  external_key varchar(50) not null comment 'Bundle external key'
+, name varchar(20) not null
+) engine=innodb comment 'Tags associated to bundles';
+
+drop table if exists bst_fields;
+create table bst_fields (
+  external_key varchar(50) not null comment 'Bundle external key'
+, name varchar(30) not null
+, value varchar(255) default null
+) engine=innodb comment 'Custom fields associated to bundles';
+
+drop table if exists bin_tags;
+create table bin_tags (
+  invoice_id char(36) not null
+, name varchar(20) not null
+) engine=innodb comment 'Tags associated to invoices';
+
+drop table if exists bin_fields;
+create table bin_fields (
+  invoice_id char(36) not null
+, name varchar(30) not null
+, value varchar(255) default null
+) engine=innodb comment 'Custom fields associated to invoices';
+
+drop table if exists bip_tags;
+create table bip_tags (
+  payment_id char(36) not null
+, name varchar(20) not null
+) engine=innodb comment 'Tags associated to payments';
+
+drop table if exists bip_fields;
+create table bip_fields (
+  payment_id char(36) not null
+, name varchar(30) not null
+, value varchar(255) default null
+) engine=innodb comment 'Custom fields associated to payments';

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 0094ca4..23209b9 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
@@ -37,17 +37,17 @@ import com.ning.billing.account.api.AccountCreationEvent;
 import com.ning.billing.account.api.AccountUserApi;
 import com.ning.billing.account.api.user.DefaultAccountCreationEvent;
 import com.ning.billing.analytics.AnalyticsTestModule;
-import com.ning.billing.analytics.BusinessSubscription;
-import com.ning.billing.analytics.BusinessSubscriptionEvent;
-import com.ning.billing.analytics.BusinessSubscriptionTransition;
+import com.ning.billing.analytics.dao.BusinessAccountSqlDao;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.analytics.MockAccount;
 import com.ning.billing.analytics.MockDuration;
 import com.ning.billing.analytics.MockPhase;
 import com.ning.billing.analytics.MockPlan;
 import com.ning.billing.analytics.MockProduct;
 import com.ning.billing.analytics.TestWithEmbeddedDB;
-import com.ning.billing.analytics.dao.BusinessAccountDao;
-import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDao;
 import com.ning.billing.catalog.MockPriceList;
 import com.ning.billing.catalog.api.Catalog;
 import com.ning.billing.catalog.api.CatalogApiException;
@@ -94,8 +94,8 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
     final Plan plan = new MockPlan("platinum-monthly", product);
     final PlanPhase phase = new MockPhase(PhaseType.EVERGREEN, plan, MockDuration.UNLIMITED(), 25.95);
 
-    private static final UUID ID = UUID.randomUUID();
-    private static final String KEY = "12345";
+    private static final Long TOTAL_ORDERING = 11L;
+    private static final String EXTERNAL_KEY = "12345";
     private static final String ACCOUNT_KEY = "pierre-12345";
     private static final Currency ACCOUNT_CURRENCY = Currency.EUR;
     private static final BigDecimal INVOICE_AMOUNT = BigDecimal.valueOf(1243.11);
@@ -122,10 +122,10 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
     private Bus bus;
 
     @Inject
-    private BusinessSubscriptionTransitionDao subscriptionDao;
+    private BusinessSubscriptionTransitionSqlDao subscriptionSqlDao;
 
     @Inject
-    private BusinessAccountDao accountDao;
+    private BusinessAccountSqlDao accountSqlDao;
 
     private SubscriptionEvent transition;
     private BusinessSubscriptionTransition expectedTransition;
@@ -165,11 +165,11 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
     }
 
     private void createSubscriptionTransitionEvent(final Account account) throws EntitlementUserApiException {
-        final SubscriptionBundle bundle = entitlementApi.createBundleForAccount(account.getId(), KEY, context);
+        final SubscriptionBundle bundle = entitlementApi.createBundleForAccount(account.getId(), EXTERNAL_KEY, context);
 
         // Verify we correctly initialized the account subsystem
         Assert.assertNotNull(bundle);
-        Assert.assertEquals(bundle.getKey(), KEY);
+        Assert.assertEquals(bundle.getKey(), EXTERNAL_KEY);
 
         // Create a subscription transition event
         final UUID subscriptionId = UUID.randomUUID();
@@ -179,7 +179,7 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
 
 
         transition = new DefaultSubscriptionEvent(new SubscriptionTransitionData(
-                ID,
+                UUID.randomUUID(),
                 subscriptionId,
                 bundle.getId(),
                 EntitlementEvent.EventType.API_USER,
@@ -194,12 +194,12 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
                 plan,
                 phase,
                 priceList,
-                1L,
+                TOTAL_ORDERING,
                 null,
                 true), null);
         expectedTransition = new BusinessSubscriptionTransition(
-                ID,
-                KEY,
+                TOTAL_ORDERING,
+                EXTERNAL_KEY,
                 ACCOUNT_KEY,
                 requestedTransitionTime,
                 BusinessSubscriptionEvent.subscriptionCreated(plan.getName(), catalog, new DateTime(), new DateTime()),
@@ -261,23 +261,23 @@ public class TestAnalyticsService extends TestWithEmbeddedDB {
             Assert.fail("Unable to start the bus or service! " + t);
         }
 
-        Assert.assertNull(accountDao.getAccount(ACCOUNT_KEY));
+        Assert.assertNull(accountSqlDao.getAccount(ACCOUNT_KEY));
 
         // Send events and wait for the async part...
         bus.post(transition);
         bus.post(accountCreationNotification);
         Thread.sleep(5000);
 
-        Assert.assertEquals(subscriptionDao.getTransitions(KEY).size(), 1);
-        Assert.assertEquals(subscriptionDao.getTransitions(KEY).get(0), expectedTransition);
+        Assert.assertEquals(subscriptionSqlDao.getTransitions(EXTERNAL_KEY).size(), 1);
+        Assert.assertEquals(subscriptionSqlDao.getTransitions(EXTERNAL_KEY).get(0), expectedTransition);
 
         // Test invoice integration - the account creation notification has triggered a BAC update
-        Assert.assertTrue(accountDao.getAccount(ACCOUNT_KEY).getTotalInvoiceBalance().compareTo(INVOICE_AMOUNT) == 0);
+        Assert.assertTrue(accountSqlDao.getAccount(ACCOUNT_KEY).getTotalInvoiceBalance().compareTo(INVOICE_AMOUNT) == 0);
 
         // Post the same invoice event again - the invoice balance shouldn't change
         bus.post(invoiceCreationNotification);
         Thread.sleep(5000);
-        Assert.assertTrue(accountDao.getAccount(ACCOUNT_KEY).getTotalInvoiceBalance().compareTo(INVOICE_AMOUNT) == 0);
+        Assert.assertTrue(accountSqlDao.getAccount(ACCOUNT_KEY).getTotalInvoiceBalance().compareTo(INVOICE_AMOUNT) == 0);
 
         // Test payment integration - the fields have already been populated, just make sure the code is exercised
         bus.post(paymentInfoNotification);

diff --git a/analytics/src/test/java/com/ning/billing/analytics/dao/TestAnalyticsDao.java b/analytics/src/test/java/com/ning/billing/analytics/dao/TestAnalyticsDao.java
index c9c31be..5d6b3a6 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/dao/TestAnalyticsDao.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/dao/TestAnalyticsDao.java
@@ -19,7 +19,6 @@ package com.ning.billing.analytics.dao;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.sql.SQLException;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 
@@ -31,10 +30,10 @@ import org.testng.Assert;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-import com.ning.billing.analytics.BusinessAccount;
-import com.ning.billing.analytics.BusinessSubscription;
-import com.ning.billing.analytics.BusinessSubscriptionEvent;
-import com.ning.billing.analytics.BusinessSubscriptionTransition;
+import com.ning.billing.analytics.model.BusinessAccount;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.analytics.MockDuration;
 import com.ning.billing.analytics.MockPhase;
 import com.ning.billing.analytics.MockPlan;
@@ -51,20 +50,19 @@ import com.ning.billing.catalog.api.PlanPhase;
 import com.ning.billing.catalog.api.Product;
 import com.ning.billing.catalog.api.ProductCategory;
 import com.ning.billing.entitlement.api.user.Subscription;
-import com.ning.billing.util.tag.Tag;
 
 public class TestAnalyticsDao extends TestWithEmbeddedDB {
-    private static final UUID EVENT_ID = UUID.randomUUID();
-    private static final String EVENT_KEY = "23456";
+    private static final Long TOTAL_ORDERING = 1L;
+    private static final String EXTERNAL_KEY = "23456";
     private static final String ACCOUNT_KEY = "pierre-143343-vcc";
 
     private final Product product = new MockProduct("platinium", "subscription", ProductCategory.BASE);
     private final Plan plan = new MockPlan("platinum-monthly", product);
     private final PlanPhase phase = new MockPhase(PhaseType.EVERGREEN, plan, MockDuration.UNLIMITED(), 25.95);
 
-    private BusinessSubscriptionTransitionDao businessSubscriptionTransitionDao;
+    private BusinessSubscriptionTransitionSqlDao businessSubscriptionTransitionSqlDao;
     private BusinessSubscriptionTransition transition;
-    private BusinessAccountDao businessAccountDao;
+    private BusinessAccountSqlDao businessAccountSqlDao;
     private BusinessAccount account;
 
     private final CatalogService catalogService = Mockito.mock(CatalogService.class);
@@ -87,48 +85,38 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
         final BusinessSubscription nextSubscription = new BusinessSubscription(null, plan.getName(), phase.getName(), Currency.USD, new DateTime(DateTimeZone.UTC), Subscription.SubscriptionState.CANCELLED, UUID.randomUUID(), UUID.randomUUID(), catalog);
         final BusinessSubscriptionEvent event = BusinessSubscriptionEvent.subscriptionCancelled(plan.getName(), catalog, requestedTimestamp, requestedTimestamp);
 
-        transition = new BusinessSubscriptionTransition(EVENT_ID, EVENT_KEY, ACCOUNT_KEY, requestedTimestamp, event, prevSubscription, nextSubscription);
+        transition = new BusinessSubscriptionTransition(TOTAL_ORDERING, EXTERNAL_KEY, ACCOUNT_KEY, requestedTimestamp, event, prevSubscription, nextSubscription);
 
         final IDBI dbi = helper.getDBI();
-        businessSubscriptionTransitionDao = dbi.onDemand(BusinessSubscriptionTransitionDao.class);
+        businessSubscriptionTransitionSqlDao = dbi.onDemand(BusinessSubscriptionTransitionSqlDao.class);
 
         // Healthcheck test to make sure MySQL is setup properly
         try {
-            businessSubscriptionTransitionDao.test();
+            businessSubscriptionTransitionSqlDao.test();
         } catch (Throwable t) {
             Assert.fail(t.toString());
         }
     }
 
     private void setupBusinessAccount() {
-        final List<Tag> tags = new ArrayList<Tag>();
-        tags.add(getMockTag("batch1"));
-        tags.add(getMockTag("great,guy"));
-        account = new BusinessAccount(ACCOUNT_KEY, BigDecimal.ONE, tags, new DateTime(DateTimeZone.UTC), BigDecimal.TEN, "ERROR_NOT_ENOUGH_FUNDS", "CreditCard", "Visa", "FRANCE");
+        account = new BusinessAccount(ACCOUNT_KEY, UUID.randomUUID().toString(), BigDecimal.ONE, new DateTime(DateTimeZone.UTC), BigDecimal.TEN, "ERROR_NOT_ENOUGH_FUNDS", "CreditCard", "Visa", "FRANCE");
 
         final IDBI dbi = helper.getDBI();
-        businessAccountDao = dbi.onDemand(BusinessAccountDao.class);
+        businessAccountSqlDao = dbi.onDemand(BusinessAccountSqlDao.class);
 
         // Healthcheck test to make sure MySQL is setup properly
         try {
-            businessAccountDao.test();
+            businessAccountSqlDao.test();
         } catch (Throwable t) {
             Assert.fail(t.toString());
         }
     }
 
-    private Tag getMockTag(final String tagDefinitionName) {
-        final Tag tag = Mockito.mock(Tag.class);
-        Mockito.when(tag.getTagDefinitionName()).thenReturn(tagDefinitionName);
-        Mockito.when(tag.toString()).thenReturn(tagDefinitionName);
-        return tag;
-    }
-
     @Test(groups = "slow")
     public void testHandleDuplicatedEvents() {
         final BusinessSubscriptionTransition transitionWithNullPrev = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
@@ -136,28 +124,28 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
                 transition.getNextSubscription()
         );
 
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPrev);
-        List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPrev);
+        List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transitionWithNullPrev);
         // Try to add the same transition, with the same UUID - we should only store one though
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPrev);
-        transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPrev);
+        transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transitionWithNullPrev);
 
         // Try now to store a look-alike transition (same fields except UUID) - we should store it this time
         final BusinessSubscriptionTransition secondTransitionWithNullPrev = new BusinessSubscriptionTransition(
-                UUID.randomUUID(),
-                transition.getKey(),
+                12L,
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 null,
                 transition.getNextSubscription()
         );
-        businessSubscriptionTransitionDao.createTransition(secondTransitionWithNullPrev);
-        transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        businessSubscriptionTransitionSqlDao.createTransition(secondTransitionWithNullPrev);
+        transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 2);
         Assert.assertTrue(transitions.contains(transitionWithNullPrev));
         Assert.assertTrue(transitions.contains(secondTransitionWithNullPrev));
@@ -166,17 +154,17 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     @Test(groups = "slow")
     public void testTransitionsWithNullPrevSubscription() {
         final BusinessSubscriptionTransition transitionWithNullPrev = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 null,
                 transition.getNextSubscription()
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPrev);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPrev);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transitionWithNullPrev);
     }
@@ -184,17 +172,17 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     @Test(groups = "slow")
     public void testTransitionsWithNullNextSubscription() {
         final BusinessSubscriptionTransition transitionWithNullNext = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 transition.getPreviousSubscription(),
                 null
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullNext);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullNext);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transitionWithNullNext);
     }
@@ -203,17 +191,17 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     public void testTransitionsWithNullFieldsInSubscription() {
         final BusinessSubscription subscriptionWithNullFields = new BusinessSubscription(null, plan.getName(), phase.getName(), Currency.USD, null, null, null, null, catalog);
         final BusinessSubscriptionTransition transitionWithNullFields = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 subscriptionWithNullFields,
                 subscriptionWithNullFields
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullFields);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullFields);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transitionWithNullFields);
     }
@@ -222,19 +210,19 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     public void testTransitionsWithNullPlanAndPhase() throws Exception {
         final BusinessSubscription subscriptionWithNullPlanAndPhase = new BusinessSubscription(null, null, null, Currency.USD, null, null, null, null, catalog);
         final BusinessSubscriptionTransition transitionWithNullPlanAndPhase = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 subscriptionWithNullPlanAndPhase,
                 subscriptionWithNullPlanAndPhase
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPlanAndPhase);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPlanAndPhase);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
-        Assert.assertEquals(transitions.get(0).getKey(), transition.getKey());
+        Assert.assertEquals(transitions.get(0).getExternalKey(), transition.getExternalKey());
         Assert.assertEquals(transitions.get(0).getRequestedTimestamp(), transition.getRequestedTimestamp());
         Assert.assertEquals(transitions.get(0).getEvent(), transition.getEvent());
         Assert.assertNull(transitions.get(0).getPreviousSubscription());
@@ -245,17 +233,17 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     public void testTransitionsWithNullPlan() throws Exception {
         final BusinessSubscription subscriptionWithNullPlan = new BusinessSubscription(null, null, phase.getName(), Currency.USD, null, null, null, null, catalog);
         final BusinessSubscriptionTransition transitionWithNullPlan = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 subscriptionWithNullPlan,
                 subscriptionWithNullPlan
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPlan);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPlan);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         // Null Plan but Phase - we don't turn the subscription into a null
         Assert.assertEquals(transitions.get(0), transitionWithNullPlan);
@@ -265,19 +253,19 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
     public void testTransitionsWithNullPhase() throws Exception {
         final BusinessSubscription subscriptionWithNullPhase = new BusinessSubscription(null, plan.getName(), null, Currency.USD, null, null, null, null, catalog);
         final BusinessSubscriptionTransition transitionWithNullPhase = new BusinessSubscriptionTransition(
-                transition.getId(),
-                transition.getKey(),
+                transition.getTotalOrdering(),
+                transition.getExternalKey(),
                 transition.getAccountKey(),
                 transition.getRequestedTimestamp(),
                 transition.getEvent(),
                 subscriptionWithNullPhase,
                 subscriptionWithNullPhase
         );
-        businessSubscriptionTransitionDao.createTransition(transitionWithNullPhase);
+        businessSubscriptionTransitionSqlDao.createTransition(transitionWithNullPhase);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
-        Assert.assertEquals(transitions.get(0).getKey(), transition.getKey());
+        Assert.assertEquals(transitions.get(0).getExternalKey(), transition.getExternalKey());
         Assert.assertEquals(transitions.get(0).getRequestedTimestamp(), transition.getRequestedTimestamp());
         Assert.assertEquals(transitions.get(0).getEvent(), transition.getEvent());
 
@@ -289,26 +277,22 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
 
     @Test(groups = "slow")
     public void testCreateAndRetrieveTransitions() {
-        businessSubscriptionTransitionDao.createTransition(transition);
+        businessSubscriptionTransitionSqlDao.createTransition(transition);
 
-        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionDao.getTransitions(EVENT_KEY);
+        final List<BusinessSubscriptionTransition> transitions = businessSubscriptionTransitionSqlDao.getTransitions(EXTERNAL_KEY);
         Assert.assertEquals(transitions.size(), 1);
         Assert.assertEquals(transitions.get(0), transition);
 
-        Assert.assertEquals(businessSubscriptionTransitionDao.getTransitions("Doesn't exist").size(), 0);
+        Assert.assertEquals(businessSubscriptionTransitionSqlDao.getTransitions("Doesn't exist").size(), 0);
     }
 
     @Test(groups = "slow")
     public void testCreateSaveAndRetrieveAccounts() {
         // Create and retrieve an account
-        businessAccountDao.createAccount(account);
-        final BusinessAccount foundAccount = businessAccountDao.getAccount(ACCOUNT_KEY);
+        businessAccountSqlDao.createAccount(account);
+        final BusinessAccount foundAccount = businessAccountSqlDao.getAccount(ACCOUNT_KEY);
         Assert.assertNotNull(foundAccount.getCreatedDt());
         Assert.assertEquals(foundAccount.getCreatedDt(), foundAccount.getUpdatedDt());
-        // Verify the joiner stuff
-        Assert.assertEquals(foundAccount.getTags().size(), 2);
-        Assert.assertEquals(foundAccount.getTags().get(0).getTagDefinitionName(), "batch1");
-        Assert.assertEquals(foundAccount.getTags().get(1).getTagDefinitionName(), "great,guy");
         // Verify the dates by backfilling them
         account.setCreatedDt(foundAccount.getCreatedDt());
         account.setUpdatedDt(foundAccount.getUpdatedDt());
@@ -318,14 +302,14 @@ public class TestAnalyticsDao extends TestWithEmbeddedDB {
         final DateTime previousUpdatedDt = account.getUpdatedDt();
         account.setBalance(BigDecimal.TEN);
         account.setPaymentMethod("PayPal");
-        businessAccountDao.saveAccount(account);
+        businessAccountSqlDao.saveAccount(account);
         // Verify the save worked as expected
-        account = businessAccountDao.getAccount(ACCOUNT_KEY);
+        account = businessAccountSqlDao.getAccount(ACCOUNT_KEY);
         Assert.assertEquals(Rounder.round(BigDecimal.TEN), account.getRoundedBalance());
         Assert.assertEquals("PayPal", account.getPaymentMethod());
         Assert.assertTrue(account.getUpdatedDt().compareTo(previousUpdatedDt) > 0);
 
         // ACCOUNT not found
-        Assert.assertNull(businessAccountDao.getAccount("Doesn't exist"));
+        Assert.assertNull(businessAccountSqlDao.getAccount("Doesn't exist"));
     }
 }

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 6a03abc..2d29f08 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestAnalyticsListener.java
@@ -27,6 +27,9 @@ import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.catalog.api.Catalog;
 import com.ning.billing.catalog.api.CatalogApiException;
 import com.ning.billing.catalog.api.CatalogService;
@@ -44,11 +47,11 @@ import com.ning.billing.entitlement.events.EntitlementEvent;
 import com.ning.billing.entitlement.events.user.ApiEventType;
 
 public class TestAnalyticsListener extends AnalyticsTestSuite {
-    private static final String KEY = "1234";
+    private static final String EXTERNAL_KEY = "1234";
     private static final String ACCOUNT_KEY = "pierre-1234";
     private final Currency CURRENCY = Currency.BRL;
 
-    private final MockBusinessSubscriptionTransitionDao dao = new MockBusinessSubscriptionTransitionDao();
+    private final MockBusinessSubscriptionTransitionSqlDao dao = new MockBusinessSubscriptionTransitionSqlDao();
     private final UUID subscriptionId = UUID.randomUUID();
     private final UUID bundleUUID = UUID.randomUUID();
     private final Product product = new MockProduct("platinium", "subscription", ProductCategory.BASE);
@@ -70,7 +73,7 @@ public class TestAnalyticsListener extends AnalyticsTestSuite {
 
     @BeforeMethod(groups = "fast")
     public void setUp() throws Exception {
-        final BusinessSubscriptionTransitionRecorder recorder = new BusinessSubscriptionTransitionRecorder(dao, catalogService, new MockEntitlementUserApi(bundleUUID, KEY), new MockAccountUserApi(ACCOUNT_KEY, CURRENCY));
+        final BusinessSubscriptionTransitionRecorder recorder = new BusinessSubscriptionTransitionRecorder(dao, catalogService, new MockEntitlementUserApi(bundleUUID, EXTERNAL_KEY), new MockAccountUserApi(ACCOUNT_KEY, CURRENCY));
         listener = new AnalyticsListener(recorder, null);
     }
 
@@ -80,59 +83,59 @@ public class TestAnalyticsListener extends AnalyticsTestSuite {
         final DateTime effectiveTransitionTime = new DateTime(DateTimeZone.UTC);
         final DateTime requestedTransitionTime = effectiveTransitionTime;
         final SubscriptionTransitionData firstTransition = createFirstSubscriptionTransition(requestedTransitionTime, effectiveTransitionTime);
-        final BusinessSubscriptionTransition firstBST = createExpectedFirstBST(firstTransition.getId(), requestedTransitionTime, effectiveTransitionTime);
+        final BusinessSubscriptionTransition firstBST = createExpectedFirstBST(firstTransition.getTotalOrdering(), requestedTransitionTime, effectiveTransitionTime);
         listener.handleSubscriptionTransitionChange(new DefaultSubscriptionEvent(firstTransition, effectiveTransitionTime));
-        Assert.assertEquals(dao.getTransitions(KEY).size(), 1);
-        Assert.assertEquals(dao.getTransitions(KEY).get(0), firstBST);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).size(), 1);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).get(0), firstBST);
 
         // Cancel it
         final DateTime effectiveCancelTransitionTime = new DateTime(DateTimeZone.UTC);
         final DateTime requestedCancelTransitionTime = effectiveCancelTransitionTime;
         final SubscriptionTransitionData cancelledSubscriptionTransition = createCancelSubscriptionTransition(requestedCancelTransitionTime, effectiveCancelTransitionTime, firstTransition.getNextState());
-        final BusinessSubscriptionTransition cancelledBST = createExpectedCancelledBST(cancelledSubscriptionTransition.getId(), requestedCancelTransitionTime, effectiveCancelTransitionTime, firstBST.getNextSubscription());
+        final BusinessSubscriptionTransition cancelledBST = createExpectedCancelledBST(cancelledSubscriptionTransition.getTotalOrdering(), requestedCancelTransitionTime, effectiveCancelTransitionTime, firstBST.getNextSubscription());
         listener.handleSubscriptionTransitionChange(new DefaultSubscriptionEvent(cancelledSubscriptionTransition, effectiveTransitionTime));
-        Assert.assertEquals(dao.getTransitions(KEY).size(), 2);
-        Assert.assertEquals(dao.getTransitions(KEY).get(1), cancelledBST);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).size(), 2);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).get(1), cancelledBST);
 
         // Recreate it
         final DateTime effectiveRecreatedTransitionTime = new DateTime(DateTimeZone.UTC);
         final DateTime requestedRecreatedTransitionTime = effectiveRecreatedTransitionTime;
         final SubscriptionTransitionData recreatedSubscriptionTransition = createRecreatedSubscriptionTransition(requestedRecreatedTransitionTime, effectiveRecreatedTransitionTime, cancelledSubscriptionTransition.getNextState());
-        final BusinessSubscriptionTransition recreatedBST = createExpectedRecreatedBST(recreatedSubscriptionTransition.getId(), requestedRecreatedTransitionTime, effectiveRecreatedTransitionTime, cancelledBST.getNextSubscription());
+        final BusinessSubscriptionTransition recreatedBST = createExpectedRecreatedBST(recreatedSubscriptionTransition.getTotalOrdering(), requestedRecreatedTransitionTime, effectiveRecreatedTransitionTime, cancelledBST.getNextSubscription());
         listener.handleSubscriptionTransitionChange(new DefaultSubscriptionEvent(recreatedSubscriptionTransition, effectiveTransitionTime));
-        Assert.assertEquals(dao.getTransitions(KEY).size(), 3);
-        Assert.assertEquals(dao.getTransitions(KEY).get(2), recreatedBST);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).size(), 3);
+        Assert.assertEquals(dao.getTransitions(EXTERNAL_KEY).get(2), recreatedBST);
 
     }
 
-    private BusinessSubscriptionTransition createExpectedFirstBST(final UUID id, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime) {
+    private BusinessSubscriptionTransition createExpectedFirstBST(final Long totalOrdering, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime) {
         final BusinessSubscriptionEvent event = BusinessSubscriptionEvent.subscriptionCreated(plan.getName(), catalog, effectiveTransitionTime, effectiveTransitionTime);
 
         final Subscription.SubscriptionState subscriptionState = Subscription.SubscriptionState.ACTIVE;
-        return createExpectedBST(id, event, requestedTransitionTime, effectiveTransitionTime, null, subscriptionState);
+        return createExpectedBST(totalOrdering, event, requestedTransitionTime, effectiveTransitionTime, null, subscriptionState);
     }
 
-    private BusinessSubscriptionTransition createExpectedCancelledBST(final UUID id, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime, final BusinessSubscription lastSubscription) {
+    private BusinessSubscriptionTransition createExpectedCancelledBST(final Long totalOrdering, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime, final BusinessSubscription lastSubscription) {
         final BusinessSubscriptionEvent event = BusinessSubscriptionEvent.subscriptionCancelled(plan.getName(), catalog, effectiveTransitionTime, effectiveTransitionTime);
-        return createExpectedBST(id, event, requestedTransitionTime, effectiveTransitionTime, lastSubscription, null);
+        return createExpectedBST(totalOrdering, event, requestedTransitionTime, effectiveTransitionTime, lastSubscription, null);
     }
 
-    private BusinessSubscriptionTransition createExpectedRecreatedBST(final UUID id, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime, final BusinessSubscription lastSubscription) {
+    private BusinessSubscriptionTransition createExpectedRecreatedBST(final Long totalOrdering, final DateTime requestedTransitionTime, final DateTime effectiveTransitionTime, final BusinessSubscription lastSubscription) {
         final BusinessSubscriptionEvent event = BusinessSubscriptionEvent.subscriptionRecreated(plan.getName(), catalog, effectiveTransitionTime, effectiveTransitionTime);
         final Subscription.SubscriptionState subscriptionState = Subscription.SubscriptionState.ACTIVE;
-        return createExpectedBST(id, event, requestedTransitionTime, effectiveTransitionTime, lastSubscription, subscriptionState);
+        return createExpectedBST(totalOrdering, event, requestedTransitionTime, effectiveTransitionTime, lastSubscription, subscriptionState);
     }
 
     private BusinessSubscriptionTransition createExpectedBST(
-            final UUID eventId,
+            final Long totalOrdering,
             final BusinessSubscriptionEvent eventType,
             final DateTime requestedTransitionTime,
             final DateTime effectiveTransitionTime,
             @Nullable final BusinessSubscription previousSubscription,
             @Nullable final Subscription.SubscriptionState nextState) {
         return new BusinessSubscriptionTransition(
-                eventId,
-                KEY,
+                totalOrdering,
+                EXTERNAL_KEY,
                 ACCOUNT_KEY,
                 requestedTransitionTime,
                 eventType,
@@ -224,4 +227,4 @@ public class TestAnalyticsListener extends AnalyticsTestSuite {
                 true
         );
     }
-}
\ No newline at end of file
+}

diff --git a/analytics/src/test/java/com/ning/billing/analytics/TestBusinessSubscriptionTransitionRecorder.java b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessSubscriptionTransitionRecorder.java
index b5cf557..531ab7b 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/TestBusinessSubscriptionTransitionRecorder.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessSubscriptionTransitionRecorder.java
@@ -26,7 +26,10 @@ import org.testng.annotations.Test;
 
 import com.ning.billing.account.api.Account;
 import com.ning.billing.account.api.AccountUserApi;
-import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionDao;
+import com.ning.billing.analytics.dao.BusinessSubscriptionTransitionSqlDao;
+import com.ning.billing.analytics.model.BusinessSubscription;
+import com.ning.billing.analytics.model.BusinessSubscriptionEvent;
+import com.ning.billing.analytics.model.BusinessSubscriptionTransition;
 import com.ning.billing.catalog.api.Catalog;
 import com.ning.billing.catalog.api.CatalogService;
 import com.ning.billing.catalog.api.Currency;
@@ -38,14 +41,14 @@ import com.ning.billing.entitlement.api.user.SubscriptionEvent;
 public class TestBusinessSubscriptionTransitionRecorder extends AnalyticsTestSuite {
     @Test(groups = "fast")
     public void testCreateAddOn() throws Exception {
-        final UUID key = UUID.randomUUID();
+        final UUID externalKey = UUID.randomUUID();
 
         // Setup the catalog
         final CatalogService catalogService = Mockito.mock(CatalogService.class);
         Mockito.when(catalogService.getFullCatalog()).thenReturn(Mockito.mock(Catalog.class));
 
         // Setup the dao
-        final BusinessSubscriptionTransitionDao dao = new MockBusinessSubscriptionTransitionDao();
+        final BusinessSubscriptionTransitionSqlDao sqlDao = new MockBusinessSubscriptionTransitionSqlDao();
         // Add a previous subscription to make sure it doesn't impact the addon
         final BusinessSubscription nextPrevSubscription = new BusinessSubscription(UUID.randomUUID().toString(),
                                                                                    UUID.randomUUID().toString(),
@@ -56,8 +59,8 @@ public class TestBusinessSubscriptionTransitionRecorder extends AnalyticsTestSui
                                                                                    UUID.randomUUID(),
                                                                                    UUID.randomUUID(),
                                                                                    catalogService.getFullCatalog());
-        dao.createTransition(new BusinessSubscriptionTransition(UUID.randomUUID(),
-                                                                key.toString(),
+        sqlDao.createTransition(new BusinessSubscriptionTransition(10L,
+                                                                externalKey.toString(),
                                                                 UUID.randomUUID().toString(),
                                                                 new DateTime(DateTimeZone.UTC),
                                                                 BusinessSubscriptionEvent.valueOf("ADD_MISC"),
@@ -66,17 +69,17 @@ public class TestBusinessSubscriptionTransitionRecorder extends AnalyticsTestSui
 
         // Setup the entitlement API
         final SubscriptionBundle bundle = Mockito.mock(SubscriptionBundle.class);
-        Mockito.when(bundle.getKey()).thenReturn(key.toString());
+        Mockito.when(bundle.getKey()).thenReturn(externalKey.toString());
         final EntitlementUserApi entitlementApi = Mockito.mock(EntitlementUserApi.class);
         Mockito.when(entitlementApi.getBundleFromId(Mockito.<UUID>any())).thenReturn(bundle);
 
         // Setup the account API
         final Account account = Mockito.mock(Account.class);
-        Mockito.when(account.getExternalKey()).thenReturn(key.toString());
+        Mockito.when(account.getExternalKey()).thenReturn(externalKey.toString());
         final AccountUserApi accountApi = Mockito.mock(AccountUserApi.class);
         Mockito.when(accountApi.getAccountById(bundle.getAccountId())).thenReturn(account);
 
-        final BusinessSubscriptionTransitionRecorder recorder = new BusinessSubscriptionTransitionRecorder(dao, catalogService, entitlementApi, accountApi);
+        final BusinessSubscriptionTransitionRecorder recorder = new BusinessSubscriptionTransitionRecorder(sqlDao, catalogService, entitlementApi, accountApi);
 
         // Create an new subscription event
         final SubscriptionEvent event = Mockito.mock(SubscriptionEvent.class);
@@ -87,10 +90,10 @@ public class TestBusinessSubscriptionTransitionRecorder extends AnalyticsTestSui
         Mockito.when(event.getSubscriptionStartDate()).thenReturn(new DateTime(DateTimeZone.UTC));
         recorder.subscriptionCreated(event);
 
-        Assert.assertEquals(dao.getTransitions(key.toString()).size(), 2);
-        final BusinessSubscriptionTransition transition = dao.getTransitions(key.toString()).get(1);
-        Assert.assertEquals(transition.getId(), event.getId());
-        Assert.assertEquals(transition.getAccountKey(), key.toString());
+        Assert.assertEquals(sqlDao.getTransitions(externalKey.toString()).size(), 2);
+        final BusinessSubscriptionTransition transition = sqlDao.getTransitions(externalKey.toString()).get(1);
+        Assert.assertEquals(transition.getTotalOrdering(), (long) event.getTotalOrdering());
+        Assert.assertEquals(transition.getAccountKey(), externalKey.toString());
         // Make sure all the prev_ columns are null
         Assert.assertNull(transition.getPreviousSubscription());
     }

diff --git a/api/src/main/java/com/ning/billing/invoice/api/InvoicePaymentApi.java b/api/src/main/java/com/ning/billing/invoice/api/InvoicePaymentApi.java
index 646a198..9c6eb69 100644
--- a/api/src/main/java/com/ning/billing/invoice/api/InvoicePaymentApi.java
+++ b/api/src/main/java/com/ning/billing/invoice/api/InvoicePaymentApi.java
@@ -44,9 +44,9 @@ public interface InvoicePaymentApi {
 
     public void notifyOfPaymentAttempt(UUID invoiceId, UUID paymentAttemptId, DateTime paymentAttemptDate, CallContext context);
 
-    public void processChargeback(UUID invoicePaymentId, BigDecimal amount, CallContext context) throws InvoiceApiException;
+    public InvoicePayment processChargeback(UUID invoicePaymentId, BigDecimal amount, CallContext context) throws InvoiceApiException;
 
-    public void processChargeback(UUID invoicePaymentId, CallContext context) throws InvoiceApiException;
+    public InvoicePayment processChargeback(UUID invoicePaymentId, CallContext context) throws InvoiceApiException;
 
     public BigDecimal getRemainingAmountPaid(UUID invoicePaymentId);
 

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
index 7316c3b..38bf0ca 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
@@ -484,6 +484,9 @@ public class AuditedEntitlementDao implements EntitlementDao {
     }
 
     private List<Subscription> buildBundleSubscriptions(final SubscriptionFactory factory, final List<Subscription> input) {
+        if (input == null || input.size() == 0) {
+            return Collections.emptyList();
+        }
         // Make sure BasePlan -- if exists-- is first
         Collections.sort(input, new Comparator<Subscription>() {
             @Override

diff --git a/entitlement/src/main/resources/com/ning/billing/entitlement/ddl.sql b/entitlement/src/main/resources/com/ning/billing/entitlement/ddl.sql
index f69a20d..ae45518 100644
--- a/entitlement/src/main/resources/com/ning/billing/entitlement/ddl.sql
+++ b/entitlement/src/main/resources/com/ning/billing/entitlement/ddl.sql
@@ -25,6 +25,8 @@ CREATE UNIQUE INDEX subscription_events_id ON subscription_events(id);
 CREATE INDEX idx_ent_1 ON subscription_events(subscription_id, is_active, effective_date);
 CREATE INDEX idx_ent_2 ON subscription_events(subscription_id, effective_date, created_date, requested_date,id);
 
+
+
 DROP TABLE IF EXISTS subscriptions;
 CREATE TABLE subscriptions (
     record_id int(11) unsigned NOT NULL AUTO_INCREMENT,
@@ -43,6 +45,7 @@ CREATE TABLE subscriptions (
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
 CREATE UNIQUE INDEX subscriptions_id ON subscriptions(id);
+CREATE INDEX subscriptions_bundle_id ON subscriptions(bundle_id);
 
 DROP TABLE IF EXISTS bundles;
 CREATE TABLE bundles (
@@ -54,4 +57,7 @@ CREATE TABLE bundles (
     last_sys_update_date datetime,
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
- CREATE UNIQUE INDEX bundles_id ON bundles(id);
+CREATE UNIQUE INDEX bundles_id ON bundles(id);
+CREATE INDEX bundles_key ON bundles(external_key);
+CREATE INDEX bundles_account ON bundles(account_id);
+

diff --git a/invoice/src/main/java/com/ning/billing/invoice/api/invoice/DefaultInvoicePaymentApi.java b/invoice/src/main/java/com/ning/billing/invoice/api/invoice/DefaultInvoicePaymentApi.java
index fca4789..a93ee28 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/api/invoice/DefaultInvoicePaymentApi.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/api/invoice/DefaultInvoicePaymentApi.java
@@ -80,16 +80,16 @@ public class DefaultInvoicePaymentApi implements InvoicePaymentApi {
     }
 
     @Override
-    public void processChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
-        dao.postChargeback(invoicePaymentId, amount, context);
+    public InvoicePayment processChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
+        return dao.postChargeback(invoicePaymentId, amount, context);
     }
 
     @Override
-    public void processChargeback(final UUID invoicePaymentId, final CallContext context) throws InvoiceApiException {
+    public InvoicePayment processChargeback(final UUID invoicePaymentId, final CallContext context) throws InvoiceApiException {
         // use the invoicePaymentId to get the amount remaining on the payment
         // (preventing charge backs totalling more than the payment)
         final BigDecimal amount = dao.getRemainingAmountPaid(invoicePaymentId);
-        processChargeback(invoicePaymentId, amount, context);
+        return processChargeback(invoicePaymentId, amount, context);
     }
 
     @Override

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
index 22ddb71..7102051 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
@@ -255,7 +255,7 @@ public class DefaultInvoiceDao implements InvoiceDao {
 
     @Override
     public InvoicePayment getInvoicePayment(final UUID paymentAttemptId) {
-        return invoicePaymentSqlDao.getInvoicePayment(paymentAttemptId);
+        return invoicePaymentSqlDao.getInvoicePayment(paymentAttemptId.toString());
     }
 
     @Override
@@ -269,18 +269,25 @@ public class DefaultInvoiceDao implements InvoiceDao {
     }
 
     @Override
-    public void postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
-        final InvoicePayment payment = invoicePaymentSqlDao.getById(invoicePaymentId.toString());
-        if (payment == null) {
-            throw new InvoiceApiException(ErrorCode.INVOICE_PAYMENT_NOT_FOUND, invoicePaymentId.toString());
-        } else {
-            if (amount.compareTo(BigDecimal.ZERO) < 0) {
-                throw new InvoiceApiException(ErrorCode.CHARGE_BACK_AMOUNT_IS_NEGATIVE);
+    public InvoicePayment postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
+        return invoicePaymentSqlDao.inTransaction(new Transaction<InvoicePayment, InvoicePaymentSqlDao>() {
+            @Override
+            public InvoicePayment inTransaction(final InvoicePaymentSqlDao transactional, final TransactionStatus status) throws Exception {
+                final InvoicePayment payment = invoicePaymentSqlDao.getById(invoicePaymentId.toString());
+                if (payment == null) {
+                    throw new InvoiceApiException(ErrorCode.INVOICE_PAYMENT_NOT_FOUND, invoicePaymentId.toString());
+                } else {
+                    if (amount.compareTo(BigDecimal.ZERO) < 0) {
+                        throw new InvoiceApiException(ErrorCode.CHARGE_BACK_AMOUNT_IS_NEGATIVE);
+                    }
+
+                    final InvoicePayment chargeBack = payment.asChargeBack(amount, context.getCreatedDate());
+                    invoicePaymentSqlDao.create(chargeBack, context);
+
+                    return chargeBack;
+                }
             }
-
-            final InvoicePayment chargeBack = payment.asChargeBack(amount, context.getCreatedDate());
-            invoicePaymentSqlDao.create(chargeBack, context);
-        }
+        });
     }
 
     @Override

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
index fe82d6c..888295a 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
@@ -61,7 +61,7 @@ public interface InvoiceDao {
 
     void removeWrittenOff(final UUID invoiceId, final CallContext context) throws TagApiException;
 
-    void postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException;
+    InvoicePayment postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException;
 
     BigDecimal getRemainingAmountPaid(final UUID invoicePaymentId);
 

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.java
index ebbdbb7..8b12f8d 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.java
@@ -78,7 +78,7 @@ public interface InvoicePaymentSqlDao extends EntitySqlDao<InvoicePayment>, Tran
     public List<InvoicePayment> getPaymentsForInvoice(@Bind("invoiceId") final String invoiceId);
 
     @SqlQuery
-    InvoicePayment getInvoicePayment(@Bind("paymentAttemptId") final UUID paymentAttemptId);
+    InvoicePayment getInvoicePayment(@Bind("paymentAttemptId") final String paymentAttemptId);
 
     @SqlUpdate
     void notifyOfPaymentAttempt(@InvoicePaymentBinder final InvoicePayment invoicePayment,

diff --git a/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java b/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
index 5793dec..601ebeb 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
@@ -59,7 +59,8 @@ public class InvoiceListener {
         try {
             //  Skip future uncancel event
             //  Skip events which are marked as not being the last one
-            if (transition.getTransitionType() == SubscriptionTransitionType.UNCANCEL
+            if (transition.getTransitionType() == SubscriptionTransitionType.UNCANCEL ||
+                    transition.getTransitionType() == SubscriptionTransitionType.MIGRATE_ENTITLEMENT
                     || transition.getRemainingEventsForUserOperation() > 0) {
                 return;
             }

diff --git a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.sql.stg b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.sql.stg
index 3773a1c..8ccf73b 100644
--- a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.sql.stg
+++ b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoicePaymentSqlDao.sql.stg
@@ -27,7 +27,7 @@ batchCreateFromTransaction() ::= <<
 getByPaymentAttemptId() ::= <<
   SELECT <invoicePaymentFields()>
   FROM invoice_payments
-  WHERE payment_id = :paymentAttemptId;
+  WHERE payment_attempt_id = :paymentAttemptId;
 >>
 
 get() ::= <<
@@ -56,7 +56,7 @@ notifyOfPaymentAttempt() ::= <<
 getInvoicePayment() ::= <<
     SELECT <invoicePaymentFields()>
     FROM invoice_payments
-    WHERE payment_id = :payment_id;
+    WHERE payment_attempt_id = :paymentAttemptId;
 >>
 
 getRecordId() ::= <<
@@ -119,4 +119,4 @@ getChargebacksByAttemptPaymentId() ::= <<
     WHERE reversed_invoice_payment_id IN
         (SELECT id FROM invoice_payments WHERE payment_attempt_id = :paymentAttemptId);
 >>
-;
\ No newline at end of file
+;

diff --git a/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql b/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
index ee67f37..7c69562 100644
--- a/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
+++ b/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
@@ -77,7 +77,7 @@ CREATE TABLE invoices (
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
 CREATE UNIQUE INDEX invoices_id ON invoices(id);
-CREATE INDEX invoices_account_id ON invoices(account_id ASC);
+CREATE INDEX invoices_account_target ON invoices(account_id ASC, target_date);
 
 DROP TABLE IF EXISTS invoice_payments;
 CREATE TABLE invoice_payments (
@@ -94,6 +94,7 @@ CREATE TABLE invoice_payments (
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
 CREATE UNIQUE INDEX invoice_payments_id ON invoice_payments(id);
+CREATE INDEX invoice_payments_attempt ON invoice_payments(payment_attempt_id);
 CREATE INDEX invoice_payments_reversals ON invoice_payments(reversed_invoice_payment_id);
 
 DROP VIEW IF EXISTS invoice_payment_summary;

diff --git a/invoice/src/test/java/com/ning/billing/invoice/api/MockInvoicePaymentApi.java b/invoice/src/test/java/com/ning/billing/invoice/api/MockInvoicePaymentApi.java
index e6349c9..d4feb57 100644
--- a/invoice/src/test/java/com/ning/billing/invoice/api/MockInvoicePaymentApi.java
+++ b/invoice/src/test/java/com/ning/billing/invoice/api/MockInvoicePaymentApi.java
@@ -102,7 +102,7 @@ public class MockInvoicePaymentApi implements InvoicePaymentApi {
     }
 
     @Override
-    public void processChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
+    public InvoicePayment processChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
         InvoicePayment existingPayment = null;
         for (final InvoicePayment payment : invoicePayments) {
             if (payment.getId() == invoicePaymentId) {
@@ -113,10 +113,12 @@ public class MockInvoicePaymentApi implements InvoicePaymentApi {
         if (existingPayment != null) {
             invoicePayments.add(existingPayment.asChargeBack(amount, DateTime.now(DateTimeZone.UTC)));
         }
+
+        return existingPayment;
     }
 
     @Override
-    public void processChargeback(final UUID invoicePaymentId, final CallContext context) throws InvoiceApiException {
+    public InvoicePayment processChargeback(final UUID invoicePaymentId, final CallContext context) throws InvoiceApiException {
         InvoicePayment existingPayment = null;
         for (final InvoicePayment payment : invoicePayments) {
             if (payment.getId() == invoicePaymentId) {
@@ -127,6 +129,8 @@ public class MockInvoicePaymentApi implements InvoicePaymentApi {
         if (existingPayment != null) {
             this.processChargeback(invoicePaymentId, existingPayment.getAmount(), context);
         }
+
+        return existingPayment;
     }
 
     @Override

diff --git a/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java b/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
index 68987f8..0564bd0 100644
--- a/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
+++ b/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
@@ -213,7 +213,7 @@ public class MockInvoiceDao implements InvoiceDao {
     }
 
     @Override
-    public void postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
+    public InvoicePayment postChargeback(final UUID invoicePaymentId, final BigDecimal amount, final CallContext context) throws InvoiceApiException {
         throw new UnsupportedOperationException();
     }
 

diff --git a/jaxrs/src/main/java/com/ning/billing/jaxrs/resources/ChargebackResource.java b/jaxrs/src/main/java/com/ning/billing/jaxrs/resources/ChargebackResource.java
index 9420e7c..8823018 100644
--- a/jaxrs/src/main/java/com/ning/billing/jaxrs/resources/ChargebackResource.java
+++ b/jaxrs/src/main/java/com/ning/billing/jaxrs/resources/ChargebackResource.java
@@ -114,20 +114,22 @@ public class ChargebackResource implements JaxrsResource {
                 }
             });
             if (attempts.size() == 0) {
-                final String error = String.format("Failed to locate succesful payment attempts for paymentId %s", paymentId);
+                final String error = String.format("Failed to locate successful payment attempts for paymentId %s", paymentId);
                 return Response.status(Response.Status.NO_CONTENT).entity(error).build();
             }
             final UUID paymentAttemptId = attempts.iterator().next().getId();
 
             final List<InvoicePayment> chargebacks = invoicePaymentApi.getChargebacksByPaymentAttemptId(paymentAttemptId);
-            final List<ChargebackJson> chargebacksJson = convertToJson(chargebacks);
-
-            final String accountId = invoicePaymentApi.getAccountIdFromInvoicePaymentId(UUID.fromString(paymentId)).toString();
-
+            if (chargebacks.size() == 0) {
+                return Response.status(Response.Status.NO_CONTENT).build();
+            }
 
+            final UUID invoicePaymentId = chargebacks.get(0).getId();
+            final String accountId = invoicePaymentApi.getAccountIdFromInvoicePaymentId(invoicePaymentId).toString();
+            final List<ChargebackJson> chargebacksJson = convertToJson(chargebacks);
             final ChargebackCollectionJson json = new ChargebackCollectionJson(accountId, chargebacksJson);
-            return Response.status(Response.Status.OK).entity(json).build();
 
+            return Response.status(Response.Status.OK).entity(json).build();
         } catch (PaymentApiException e) {
             final String error = String.format("Failed to locate payment attempt for payment id %s", paymentId);
             return Response.status(Response.Status.NO_CONTENT).entity(error).build();
@@ -145,15 +147,36 @@ public class ChargebackResource implements JaxrsResource {
                                      @HeaderParam(HDR_REASON) final String reason,
                                      @HeaderParam(HDR_COMMENT) final String comment) {
         try {
-            invoicePaymentApi.processChargeback(UUID.fromString(json.getPaymentId()), json.getChargebackAmount(),
-                                                context.createContext(createdBy, reason, comment));
-            return uriBuilder.buildResponse(ChargebackResource.class, "getChargeback", json.getPaymentId());
+            final Payment payment = paymentApi.getPayment(UUID.fromString(json.getPaymentId()));
+            final Collection<PaymentAttempt> attempts = Collections2.filter(payment.getAttempts(), new Predicate<PaymentAttempt>() {
+                @Override
+                public boolean apply(final PaymentAttempt input) {
+                    return input.getPaymentStatus() == PaymentStatus.SUCCESS;
+                }
+            });
+            if (attempts.size() == 0) {
+                final String error = String.format("Failed to locate successful payment attempts for paymentId %s", json.getPaymentId());
+                return Response.status(Response.Status.NO_CONTENT).entity(error).build();
+            }
+
+            final UUID paymentAttemptId = attempts.iterator().next().getId();
+            final InvoicePayment invoicePayment = invoicePaymentApi.getInvoicePayment(paymentAttemptId);
+            if (invoicePayment == null) {
+                final String error = String.format("Failed to locate invoice payment for paymentAttemptId %s", paymentAttemptId);
+                return Response.status(Response.Status.NO_CONTENT).entity(error).build();
+            }
+
+            final InvoicePayment chargeBack = invoicePaymentApi.processChargeback(invoicePayment.getId(), json.getChargebackAmount(),
+                                                                                  context.createContext(createdBy, reason, comment));
+            return uriBuilder.buildResponse(ChargebackResource.class, "getChargeback", chargeBack.getId());
         } catch (InvoiceApiException e) {
             final String error = String.format("Failed to create chargeback %s", json);
             log.info(error, e);
             return Response.status(Response.Status.BAD_REQUEST).entity(error).build();
         } catch (IllegalArgumentException e) {
             return Response.status(Response.Status.BAD_REQUEST).entity(e.getMessage()).build();
+        } catch (PaymentApiException e) {
+            return Response.status(Response.Status.BAD_REQUEST).entity(e.getMessage()).build();
         }
     }
 

diff --git a/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java b/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
index 78e1c99..20e30b5 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
@@ -79,16 +79,16 @@ public class PaymentProcessor extends ProcessorBase {
 
     @Inject
     public PaymentProcessor(final PaymentProviderPluginRegistry pluginRegistry,
-                            final AccountUserApi accountUserApi,
-                            final InvoicePaymentApi invoicePaymentApi,
-                            final FailedPaymentRetryServiceScheduler failedPaymentRetryService,
-                            final PluginFailureRetryServiceScheduler pluginFailureRetryService,
-                            final PaymentDao paymentDao,
-                            final Bus eventBus,
-                            final Clock clock,
-                            final GlobalLocker locker,
-                            @Named(PLUGIN_EXECUTOR_NAMED) final ExecutorService executor,
-                            final CallContextFactory factory) {
+            final AccountUserApi accountUserApi,
+            final InvoicePaymentApi invoicePaymentApi,
+            final FailedPaymentRetryServiceScheduler failedPaymentRetryService,
+            final PluginFailureRetryServiceScheduler pluginFailureRetryService,
+            final PaymentDao paymentDao,
+            final Bus eventBus,
+            final Clock clock,
+            final GlobalLocker locker,
+            @Named(PLUGIN_EXECUTOR_NAMED) final ExecutorService executor,
+            final CallContextFactory factory) {
         super(pluginRegistry, accountUserApi, eventBus, paymentDao, locker, executor);
         this.invoicePaymentApi = invoicePaymentApi;
         this.failedPaymentRetryService = failedPaymentRetryService;
@@ -131,7 +131,7 @@ public class PaymentProcessor extends ProcessorBase {
     }
 
     public Payment createPayment(final String accountKey, final UUID invoiceId, final BigDecimal inputAmount, final CallContext context, final boolean isInstantPayment)
-            throws PaymentApiException {
+    throws PaymentApiException {
         try {
             final Account account = accountUserApi.getAccountByKey(accountKey);
             return createPayment(account, invoiceId, inputAmount, context, isInstantPayment);
@@ -141,28 +141,28 @@ public class PaymentProcessor extends ProcessorBase {
     }
 
     public Payment createPayment(final Account account, final UUID invoiceId, final BigDecimal inputAmount, final CallContext context, final boolean isInstantPayment)
-            throws PaymentApiException {
+    throws PaymentApiException {
 
         final PaymentPluginApi plugin = getPaymentProviderPlugin(account);
 
         try {
             return paymentPluginDispatcher.dispatchWithAccountLock(new CallableWithAccountLock<Payment>(locker,
-                                                                                                        account.getExternalKey(),
-                                                                                                        new WithAccountLockCallback<Payment>() {
-
-                                                                                                            @Override
-                                                                                                            public Payment doOperation() throws PaymentApiException {
-                                                                                                                final Invoice invoice = invoicePaymentApi.getInvoice(invoiceId);
-
-                                                                                                                if (invoice.isMigrationInvoice()) {
-                                                                                                                    log.error("Received invoice for payment that is a migration invoice - don't know how to handle those yet: {}", invoice);
-                                                                                                                    return null;
-                                                                                                                }
-
-                                                                                                                final BigDecimal requestedAmount = getAndValidatePaymentAmount(invoice, inputAmount, isInstantPayment);
-                                                                                                                return processNewPaymentWithAccountLocked(plugin, account, invoice, requestedAmount, isInstantPayment, context);
-                                                                                                            }
-                                                                                                        }));
+                    account.getExternalKey(),
+                    new WithAccountLockCallback<Payment>() {
+
+                @Override
+                public Payment doOperation() throws PaymentApiException {
+                    final Invoice invoice = invoicePaymentApi.getInvoice(invoiceId);
+
+                    if (invoice.isMigrationInvoice()) {
+                        log.error("Received invoice for payment that is a migration invoice - don't know how to handle those yet: {}", invoice);
+                        return null;
+                    }
+
+                    final BigDecimal requestedAmount = getAndValidatePaymentAmount(invoice, inputAmount, isInstantPayment);
+                    return processNewPaymentWithAccountLocked(plugin, account, invoice, requestedAmount, isInstantPayment, context);
+                }
+            }));
         } catch (TimeoutException e) {
             if (isInstantPayment) {
                 throw new PaymentApiException(ErrorCode.PAYMENT_PLUGIN_TIMEOUT, account.getId(), invoiceId);
@@ -179,7 +179,7 @@ public class PaymentProcessor extends ProcessorBase {
 
 
     private BigDecimal getAndValidatePaymentAmount(final Invoice invoice, final BigDecimal inputAmount, final boolean isInstantPayment)
-            throws PaymentApiException {
+    throws PaymentApiException {
 
         if (invoice.getBalance().compareTo(BigDecimal.ZERO) <= 0) {
             throw new PaymentApiException(ErrorCode.PAYMENT_NULL_INVOICE, invoice.getId());
@@ -188,7 +188,7 @@ public class PaymentProcessor extends ProcessorBase {
                 inputAmount != null &&
                 invoice.getBalance().compareTo(inputAmount) < 0) {
             throw new PaymentApiException(ErrorCode.PAYMENT_AMOUNT_DENIED,
-                                          invoice.getId(), inputAmount.floatValue(), invoice.getBalance().floatValue());
+                    invoice.getId(), inputAmount.floatValue(), invoice.getBalance().floatValue());
         }
         return inputAmount != null ? inputAmount : invoice.getBalance();
     }
@@ -217,39 +217,39 @@ public class PaymentProcessor extends ProcessorBase {
             final CallContext context = factory.createCallContext("PaymentRetry", CallOrigin.INTERNAL, UserType.SYSTEM);
 
             voidPluginDispatcher.dispatchWithAccountLock(new CallableWithAccountLock<Void>(locker,
-                                                                                           account.getExternalKey(),
-                                                                                           new WithAccountLockCallback<Void>() {
-
-                                                                                               @Override
-                                                                                               public Void doOperation() throws PaymentApiException {
-
-                                                                                                   // Fetch gain with account lock this time
-                                                                                                   final PaymentModelDao payment = paymentDao.getPayment(paymentId);
-                                                                                                   boolean foundExpectedState = false;
-                                                                                                   for (final PaymentStatus cur : expectedPaymentStates) {
-                                                                                                       if (payment.getPaymentStatus() == cur) {
-                                                                                                           foundExpectedState = true;
-                                                                                                           break;
-                                                                                                       }
-                                                                                                   }
-                                                                                                   if (!foundExpectedState) {
-                                                                                                       log.info("Aborted retry for payment {} because it is {} state", paymentId, payment.getPaymentStatus());
-                                                                                                       return null;
-                                                                                                   }
-
-                                                                                                   final Invoice invoice = invoicePaymentApi.getInvoice(payment.getInvoiceId());
-                                                                                                   if (invoice.isMigrationInvoice()) {
-                                                                                                       return null;
-                                                                                                   }
-                                                                                                   if (invoice.getBalance().compareTo(BigDecimal.ZERO) <= 0) {
-                                                                                                       log.info("Aborted retry for payment {} because invoice has been paid", paymentId);
-                                                                                                       return null;
-                                                                                                   }
-                                                                                                   processRetryPaymentWithAccountLocked(plugin, account, invoice, payment, invoice.getBalance(), context);
-                                                                                                   return null;
-
-                                                                                               }
-                                                                                           }));
+                    account.getExternalKey(),
+                    new WithAccountLockCallback<Void>() {
+
+                @Override
+                public Void doOperation() throws PaymentApiException {
+
+                    // Fetch gain with account lock this time
+                    final PaymentModelDao payment = paymentDao.getPayment(paymentId);
+                    boolean foundExpectedState = false;
+                    for (final PaymentStatus cur : expectedPaymentStates) {
+                        if (payment.getPaymentStatus() == cur) {
+                            foundExpectedState = true;
+                            break;
+                        }
+                    }
+                    if (!foundExpectedState) {
+                        log.info("Aborted retry for payment {} because it is {} state", paymentId, payment.getPaymentStatus());
+                        return null;
+                    }
+
+                    final Invoice invoice = invoicePaymentApi.getInvoice(payment.getInvoiceId());
+                    if (invoice.isMigrationInvoice()) {
+                        return null;
+                    }
+                    if (invoice.getBalance().compareTo(BigDecimal.ZERO) <= 0) {
+                        log.info("Aborted retry for payment {} because invoice has been paid", paymentId);
+                        return null;
+                    }
+                    processRetryPaymentWithAccountLocked(plugin, account, invoice, payment, invoice.getBalance(), context);
+                    return null;
+
+                }
+            }));
         } catch (AccountApiException e) {
             log.error(String.format("Failed to retry payment for paymentId %s", paymentId), e);
         } catch (PaymentApiException e) {
@@ -261,7 +261,7 @@ public class PaymentProcessor extends ProcessorBase {
     }
 
     private Payment processNewPaymentWithAccountLocked(final PaymentPluginApi plugin, final Account account, final Invoice invoice,
-                                                       final BigDecimal requestedAmount, final boolean isInstantPayment, final CallContext context) throws PaymentApiException {
+            final BigDecimal requestedAmount, final boolean isInstantPayment, final CallContext context) throws PaymentApiException {
 
         final boolean scheduleRetryForPayment = !isInstantPayment;
         final PaymentModelDao payment = new PaymentModelDao(account.getId(), invoice.getId(), requestedAmount.setScale(2, RoundingMode.HALF_EVEN), invoice.getCurrency(), invoice.getTargetDate());
@@ -272,7 +272,7 @@ public class PaymentProcessor extends ProcessorBase {
     }
 
     private Payment processRetryPaymentWithAccountLocked(final PaymentPluginApi plugin, final Account account, final Invoice invoice, final PaymentModelDao payment,
-                                                         final BigDecimal requestedAmount, final CallContext context) throws PaymentApiException {
+            final BigDecimal requestedAmount, final CallContext context) throws PaymentApiException {
         final boolean scheduleRetryForPayment = true;
         final PaymentAttemptModelDao attempt = new PaymentAttemptModelDao(account.getId(), invoice.getId(), payment.getId(), clock.getUTCNow(), requestedAmount);
         paymentDao.insertNewAttemptForPayment(payment.getId(), attempt, scheduleRetryForPayment, context);
@@ -281,7 +281,7 @@ public class PaymentProcessor extends ProcessorBase {
 
 
     private Payment processPaymentWithAccountLocked(final PaymentPluginApi plugin, final Account account, final Invoice invoice,
-                                                    final PaymentModelDao paymentInput, final PaymentAttemptModelDao attemptInput, final boolean isInstantPayment, final CallContext context) throws PaymentApiException {
+            final PaymentModelDao paymentInput, final PaymentAttemptModelDao attemptInput, final boolean isInstantPayment, final CallContext context) throws PaymentApiException {
 
         BusEvent event = null;
         List<PaymentAttemptModelDao> allAttempts = null;
@@ -292,52 +292,52 @@ public class PaymentProcessor extends ProcessorBase {
 
             final PaymentInfoPlugin paymentPluginInfo = plugin.processPayment(account.getExternalKey(), paymentInput.getId(), attemptInput.getRequestedAmount());
             switch (paymentPluginInfo.getStatus()) {
-                case PROCESSED:
-                    // Update Payment/PaymentAttempt status
-                    paymentStatus = PaymentStatus.SUCCESS;
-                    paymentDao.updateStatusForPaymentWithAttempt(paymentInput.getId(), paymentStatus, null, attemptInput.getId(), context);
-
-                    // Fetch latest objects
+            case PROCESSED:
+                // Update Payment/PaymentAttempt status
+                paymentStatus = PaymentStatus.SUCCESS;
+                paymentDao.updateStatusForPaymentWithAttempt(paymentInput.getId(), paymentStatus, null, attemptInput.getId(), context);
+
+                // Fetch latest objects
+                allAttempts = paymentDao.getAttemptsForPayment(paymentInput.getId());
+                lastAttempt = allAttempts.get(allAttempts.size() - 1);
+                payment = paymentDao.getPayment(paymentInput.getId());
+
+                invoicePaymentApi.notifyOfPaymentAttempt(invoice.getId(),
+                        paymentStatus == PaymentStatus.SUCCESS ? payment.getAmount() : null,
+                                paymentStatus == PaymentStatus.SUCCESS ? payment.getCurrency() : null,
+                                        lastAttempt.getId(),
+                                        lastAttempt.getEffectiveDate(),
+                                        context);
+
+                // Create Bus event
+                event = new DefaultPaymentInfoEvent(account.getId(),
+                        invoice.getId(), payment.getId(), payment.getAmount(), payment.getPaymentNumber(), paymentStatus, context.getUserToken(), payment.getEffectiveDate());
+                break;
+
+            case ERROR:
+                // Schedule if non instant payment and max attempt for retry not reached yet
+                if (!isInstantPayment) {
                     allAttempts = paymentDao.getAttemptsForPayment(paymentInput.getId());
-                    lastAttempt = allAttempts.get(allAttempts.size() - 1);
-                    payment = paymentDao.getPayment(paymentInput.getId());
-
-                    invoicePaymentApi.notifyOfPaymentAttempt(invoice.getId(),
-                                                             paymentStatus == PaymentStatus.SUCCESS ? payment.getAmount() : null,
-                                                             paymentStatus == PaymentStatus.SUCCESS ? payment.getCurrency() : null,
-                                                             lastAttempt.getId(),
-                                                             lastAttempt.getEffectiveDate(),
-                                                             context);
-
-                    // Create Bus event
-                    event = new DefaultPaymentInfoEvent(account.getId(),
-                                                        invoice.getId(), payment.getId(), payment.getAmount(), payment.getPaymentNumber(), paymentStatus, context.getUserToken(), payment.getEffectiveDate());
-                    break;
-
-                case ERROR:
-                    // Schedule if non instant payment and max attempt for retry not reached yet
-                    if (!isInstantPayment) {
-                        allAttempts = paymentDao.getAttemptsForPayment(paymentInput.getId());
-                        final int retryAttempt = getNumberAttemptsInState(paymentInput.getId(), allAttempts,
-                                                                          PaymentStatus.UNKNOWN, PaymentStatus.PAYMENT_FAILURE);
-                        final boolean isScheduledForRetry = failedPaymentRetryService.scheduleRetry(paymentInput.getId(), retryAttempt);
-                        paymentStatus = isScheduledForRetry ? PaymentStatus.PAYMENT_FAILURE : PaymentStatus.PAYMENT_FAILURE_ABORTED;
-                    } else {
-                        paymentStatus = PaymentStatus.PAYMENT_FAILURE_ABORTED;
-                    }
+                    final int retryAttempt = getNumberAttemptsInState(paymentInput.getId(), allAttempts,
+                            PaymentStatus.UNKNOWN, PaymentStatus.PAYMENT_FAILURE);
+                    final boolean isScheduledForRetry = failedPaymentRetryService.scheduleRetry(paymentInput.getId(), retryAttempt);
+                    paymentStatus = isScheduledForRetry ? PaymentStatus.PAYMENT_FAILURE : PaymentStatus.PAYMENT_FAILURE_ABORTED;
+                } else {
+                    paymentStatus = PaymentStatus.PAYMENT_FAILURE_ABORTED;
+                }
 
-                    paymentDao.updateStatusForPaymentWithAttempt(paymentInput.getId(), paymentStatus, paymentPluginInfo.getGatewayError(), attemptInput.getId(), context);
+                paymentDao.updateStatusForPaymentWithAttempt(paymentInput.getId(), paymentStatus, paymentPluginInfo.getGatewayError(), attemptInput.getId(), context);
 
-                    log.info(String.format("Could not process payment for account %s, invoice %s, error = %s",
-                                           account.getId(), invoice.getId(), paymentPluginInfo.getGatewayError()));
+                log.info(String.format("Could not process payment for account %s, invoice %s, error = %s",
+                        account.getId(), invoice.getId(), paymentPluginInfo.getGatewayError()));
 
-                    event = new DefaultPaymentErrorEvent(account.getId(), invoice.getId(), paymentInput.getId(), paymentPluginInfo.getGatewayError(), context.getUserToken());
-                    throw new PaymentApiException(ErrorCode.PAYMENT_CREATE_PAYMENT, account.getId(), paymentPluginInfo.getGatewayError());
+                event = new DefaultPaymentErrorEvent(account.getId(), invoice.getId(), paymentInput.getId(), paymentPluginInfo.getGatewayError(), context.getUserToken());
+                throw new PaymentApiException(ErrorCode.PAYMENT_CREATE_PAYMENT, account.getId(), paymentPluginInfo.getGatewayError());
 
-                default:
-                    final String formatError = String.format("Plugin return status %s for payment %s", paymentPluginInfo.getStatus(), paymentInput.getId());
-                    // This caught right below as a retryable Plugin failure
-                    throw new PaymentPluginApiException("", formatError);
+            default:
+                final String formatError = String.format("Plugin return status %s for payment %s", paymentPluginInfo.getStatus(), paymentInput.getId());
+                // This caught right below as a retryable Plugin failure
+                throw new PaymentPluginApiException("", formatError);
             }
 
         } catch (PaymentPluginApiException e) {

diff --git a/payment/src/main/resources/com/ning/billing/payment/ddl.sql b/payment/src/main/resources/com/ning/billing/payment/ddl.sql
index ba4ff80..b8fa010 100644
--- a/payment/src/main/resources/com/ning/billing/payment/ddl.sql
+++ b/payment/src/main/resources/com/ning/billing/payment/ddl.sql
@@ -17,6 +17,8 @@ CREATE TABLE payments (
     PRIMARY KEY (record_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
 CREATE UNIQUE INDEX payments_id ON payments(id);
+CREATE INDEX payments_inv ON payments(invoice_id);
+CREATE INDEX payments_accnt ON payments(account_id);
 
 DROP TABLE IF EXISTS payment_history; 
 CREATE TABLE payment_history (
@@ -54,6 +56,7 @@ CREATE TABLE payment_attempts (
     PRIMARY KEY (record_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
 CREATE UNIQUE INDEX payment_attempts_id ON payment_attempts(id);
+CREATE INDEX payment_attempts_payment ON payment_attempts(payment_id);
 
 
 DROP TABLE IF EXISTS payment_attempt_history;
@@ -89,6 +92,7 @@ CREATE TABLE payment_methods (
     PRIMARY KEY (record_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
 CREATE UNIQUE INDEX payment_methods_id ON payment_methods(id);
+CREATE INDEX payment_methods_active_accnt ON payment_methods(is_active, account_id);
 
 
 DROP TABLE IF EXISTS payment_method_history;

diff --git a/server/src/test/java/com/ning/billing/jaxrs/TestChargeback.java b/server/src/test/java/com/ning/billing/jaxrs/TestChargeback.java
index 8627a0d..9dc8d51 100644
--- a/server/src/test/java/com/ning/billing/jaxrs/TestChargeback.java
+++ b/server/src/test/java/com/ning/billing/jaxrs/TestChargeback.java
@@ -18,40 +18,42 @@ package com.ning.billing.jaxrs;
 
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.util.List;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.testng.Assert;
-import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
-import com.ning.billing.invoice.api.InvoicePayment;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.ImmutableMap;
+import com.ning.billing.catalog.api.BillingPeriod;
+import com.ning.billing.catalog.api.ProductCategory;
+import com.ning.billing.jaxrs.json.AccountJson;
+import com.ning.billing.jaxrs.json.BundleJsonNoSubscriptions;
 import com.ning.billing.jaxrs.json.ChargebackCollectionJson;
 import com.ning.billing.jaxrs.json.ChargebackJson;
+import com.ning.billing.jaxrs.json.InvoiceJsonSimple;
+import com.ning.billing.jaxrs.json.PaymentJsonSimple;
+import com.ning.billing.jaxrs.json.SubscriptionJsonNoEvents;
 import com.ning.billing.jaxrs.resources.JaxrsResource;
 import com.ning.http.client.Response;
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
 
 public class TestChargeback extends TestJaxrsBase {
-    private final String accountId = UUID.randomUUID().toString();
-
-    @BeforeMethod(groups = "slow")
-    public void setUp() throws Exception {
-        final InvoicePayment invoicePayment = createInvoicePayment();
-    }
-
-    @Test(groups = "slow", enabled = false)
+    @Test(groups = "slow")
     public void testAddChargeback() throws Exception {
-        final ChargebackJson input = new ChargebackJson(new DateTime(DateTimeZone.UTC), new DateTime(DateTimeZone.UTC),
-                                                        BigDecimal.TEN, UUID.randomUUID().toString(), UUID.randomUUID().toString());
+        final PaymentJsonSimple payment = createInvoicePayment();
+        final ChargebackJson input = new ChargebackJson(null, null, BigDecimal.TEN, payment.getPaymentId(), null);
         final String jsonInput = mapper.writeValueAsString(input);
 
         // Create the chargeback
         Response response = doPost(JaxrsResource.CHARGEBACKS_PATH, jsonInput, DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
-        assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.OK.getStatusCode(), response.getResponseBody());
+        assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.CREATED.getStatusCode(), response.getResponseBody());
 
         // Find the chargeback by location
         final String location = response.getHeader("Location");
@@ -60,11 +62,11 @@ public class TestChargeback extends TestJaxrsBase {
         verifySingleChargebackResponse(response, input);
 
         // Find the chargeback by account
-        response = doGet(JaxrsResource.CHARGEBACKS_PATH + "/accounts/" + accountId, DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
+        response = doGet(JaxrsResource.CHARGEBACKS_PATH + "/accounts/" + payment.getAccountId(), DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
         verifyCollectionChargebackResponse(response, input);
 
         // Find the chargeback by payment
-        response = doGet(JaxrsResource.CHARGEBACKS_PATH + "/payments/" + input.getPaymentId(), DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
+        response = doGet(JaxrsResource.CHARGEBACKS_PATH + "/payments/" + payment.getPaymentId(), DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
         verifyCollectionChargebackResponse(response, input);
     }
 
@@ -72,13 +74,13 @@ public class TestChargeback extends TestJaxrsBase {
         assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.OK.getStatusCode());
         final ChargebackCollectionJson objFromJson = mapper.readValue(response.getResponseBody(), ChargebackCollectionJson.class);
         assertEquals(objFromJson.getChargebacks().size(), 1);
-        assertEquals(objFromJson.getChargebacks().get(0), input);
+        assertTrue(objFromJson.getChargebacks().get(0).getChargebackAmount().compareTo(input.getChargebackAmount()) == 0);
     }
 
     private void verifySingleChargebackResponse(final Response response, final ChargebackJson input) throws IOException {
         assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.OK.getStatusCode());
         final ChargebackJson objFromJson = mapper.readValue(response.getResponseBody(), ChargebackJson.class);
-        assertEquals(objFromJson, input);
+        assertTrue(objFromJson.getChargebackAmount().compareTo(input.getChargebackAmount()) == 0);
     }
 
     @Test(groups = "slow")
@@ -122,8 +124,50 @@ public class TestChargeback extends TestJaxrsBase {
         //assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.NO_CONTENT.getStatusCode(), response.getResponseBody());
     }
 
-    private InvoicePayment createInvoicePayment() {
-        // TODO - blocked on payment resource
-        return null;
+    private PaymentJsonSimple createInvoicePayment() throws Exception {
+        final InvoiceJsonSimple invoice = createInvoice();
+        return getPayment(invoice);
+    }
+
+    private InvoiceJsonSimple createInvoice() throws Exception {
+        // Create account
+        final AccountJson accountJson = createAccountWithDefaultPaymentMethod(UUID.randomUUID().toString(), UUID.randomUUID().toString(), "nohup@yahoo.com");
+
+        // Create bundle
+        final BundleJsonNoSubscriptions bundleJson = createBundle(accountJson.getAccountId(), UUID.randomUUID().toString());
+        assertNotNull(bundleJson);
+
+        // Create subscription
+        final SubscriptionJsonNoEvents subscriptionJson = createSubscription(bundleJson.getBundleId(), "Shotgun", ProductCategory.BASE.toString(), BillingPeriod.MONTHLY.toString(), true);
+        assertNotNull(subscriptionJson);
+
+        // Move after the trial period to trigger an invoice with a non-zero invoice item
+        clock.addDays(32);
+        crappyWaitForLackOfProperSynchonization();
+
+        // Retrieve the invoice
+        final Response response = doGet(JaxrsResource.INVOICES_PATH, ImmutableMap.<String, String>of(JaxrsResource.QUERY_ACCOUNT_ID, accountJson.getAccountId()), DEFAULT_HTTP_TIMEOUT_SEC);
+        Assert.assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.OK.getStatusCode());
+        final String baseJson = response.getResponseBody();
+        final List<InvoiceJsonSimple> objFromJson = mapper.readValue(baseJson, new TypeReference<List<InvoiceJsonSimple>>() {});
+        assertNotNull(objFromJson);
+        // We should have two invoices, one for the trial (zero dollar amount) and one for the first month
+        assertEquals(objFromJson.size(), 2);
+        assertTrue(objFromJson.get(1).getAmount().doubleValue() > 0);
+
+        return objFromJson.get(1);
+    }
+
+    private PaymentJsonSimple getPayment(final InvoiceJsonSimple invoice) throws IOException {
+        final String uri = JaxrsResource.INVOICES_PATH + "/" + invoice.getInvoiceId() + "/" + JaxrsResource.PAYMENTS;
+        final Response response = doGet(uri, DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
+
+        Assert.assertEquals(response.getStatusCode(), javax.ws.rs.core.Response.Status.OK.getStatusCode());
+        final String baseJson = response.getResponseBody();
+        final List<PaymentJsonSimple> objFromJson = mapper.readValue(baseJson, new TypeReference<List<PaymentJsonSimple>>() {});
+        assertNotNull(objFromJson);
+        assertEquals(objFromJson.size(), 1);
+
+        return objFromJson.get(0);
     }
 }

diff --git a/util/pom.xml b/util/pom.xml
index fed230a..1b55996 100644
--- a/util/pom.xml
+++ b/util/pom.xml
@@ -65,6 +65,11 @@
             <artifactId>joda-time</artifactId>
         </dependency>
         <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
             <groupId>org.testng</groupId>
             <artifactId>testng</artifactId>
             <scope>test</scope>

diff --git a/util/src/main/resources/com/ning/billing/util/bus/dao/PersistentBusSqlDao.sql.stg b/util/src/main/resources/com/ning/billing/util/bus/dao/PersistentBusSqlDao.sql.stg
index 0cbd4ec..61e991b 100644
--- a/util/src/main/resources/com/ning/billing/util/bus/dao/PersistentBusSqlDao.sql.stg
+++ b/util/src/main/resources/com/ning/billing/util/bus/dao/PersistentBusSqlDao.sql.stg
@@ -14,6 +14,7 @@ getNextBusEventEntry() ::= <<
     where
       processing_state != 'PROCESSED'
       and processing_state != 'REMOVED'
+      and creating_owner = :owner
       and (processing_owner IS NULL OR processing_available_date \<= :now)
     order by
       record_id asc

diff --git a/util/src/main/resources/com/ning/billing/util/ddl.sql b/util/src/main/resources/com/ning/billing/util/ddl.sql
index c61cd30..a0e7dad 100644
--- a/util/src/main/resources/com/ning/billing/util/ddl.sql
+++ b/util/src/main/resources/com/ning/billing/util/ddl.sql
@@ -113,7 +113,7 @@ CREATE TABLE notifications (
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
 CREATE UNIQUE INDEX notifications_id ON notifications(id);
-CREATE INDEX  `idx_comp_where` ON notifications (`effective_date`, `queue_name`, `processing_state`,`processing_owner`,`processing_available_date`);
+CREATE INDEX  `idx_comp_where` ON notifications (`effective_date`, `queue_name`, `processing_state`,`creating_owner`,`processing_owner`,`processing_available_date`);
 CREATE INDEX  `idx_update` ON notifications (`processing_state`,`processing_owner`,`processing_available_date`);
 CREATE INDEX  `idx_get_ready` ON notifications (`effective_date`,`created_date`,`id`);
 
@@ -154,7 +154,7 @@ CREATE TABLE bus_events (
     processing_state varchar(14) DEFAULT 'AVAILABLE',
     PRIMARY KEY(record_id)
 ) ENGINE=innodb;
-CREATE INDEX  `idx_bus_where` ON bus_events (`processing_state`,`processing_owner`,`processing_available_date`);
+CREATE INDEX  `idx_bus_where` ON bus_events (`processing_state`,`processing_owner`,`creating_owner`,`processing_available_date`);
 
 DROP TABLE IF EXISTS claimed_bus_events;
 CREATE TABLE claimed_bus_events (

diff --git a/util/src/main/resources/com/ning/billing/util/notificationq/dao/NotificationSqlDao.sql.stg b/util/src/main/resources/com/ning/billing/util/notificationq/dao/NotificationSqlDao.sql.stg
index 13ff029..8eb289a 100644
--- a/util/src/main/resources/com/ning/billing/util/notificationq/dao/NotificationSqlDao.sql.stg
+++ b/util/src/main/resources/com/ning/billing/util/notificationq/dao/NotificationSqlDao.sql.stg
@@ -19,6 +19,7 @@ getReadyNotifications() ::= <<
       and queue_name = :queueName
       and processing_state != 'PROCESSED'
       and processing_state != 'REMOVED'
+      and creating_owner = :owner
       and (processing_owner IS NULL OR processing_available_date \<= :now)
     order by
       effective_date asc