ducky/subscriptions
2015-06-14
Child:61c4ce5850da
ducky/subscriptions/subscription_postgres.go
Implement PostgreSQL support, drop subscription IDs. Create a Postgres object that wraps database/sql, so we can attach methods to it and fulfill interfaces. Create a postgres_init.sql script that will create the subscriptions table in a PostgreSQL database. Make our period type fulfill the driver.Valuer and driver.Scanner types, so it can be stored in and retrieved from SQL databases. Create a SubscriptionStats type, and add a method to our subscriptionStore interface that will allow us to retrieve current stats about the Subscriptions it is storing. Deprecated the ID property of our Subscription type, and use the Subscription.UserID property instead as our primary key. Subscriptions should be unique per user and we generally will want to access Subscriptions in the context of the User they belong to, so the UserID is a better primary key. This also means we removed the getSubscriptionByUserID method (and implementations) from our subscriptionStore, as getSubscriptions now fills that role. Implement our getSubscriptionStats method in the memstore. Implement the subscriptionStore interface on our new Postgres type. Run the subscription store tests on our Postgres type, as well, if the PG_TEST_DB environment variable is set. Round all our timestamps in our tests to the nearest millisecond, as Postgres silently truncates all timestamps to the nearest millisecond, and it was causing false test failures. Remove the tests for our getSubscriptionStoreByUser method, as that was removed.
1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000 1.2 +++ b/subscription_postgres.go Sun Jun 14 02:48:08 2015 -0400 1.3 @@ -0,0 +1,220 @@ 1.4 +package subscriptions 1.5 + 1.6 +import ( 1.7 + "database/sql" 1.8 + "time" 1.9 + 1.10 + "code.secondbit.org/uuid.hg" 1.11 + "github.com/lib/pq" 1.12 + "github.com/secondbit/pan" 1.13 +) 1.14 + 1.15 +// GetSQLTableName fulfills the pan.SQLTableNamer interface, allowing 1.16 +// us to manipulate Subscriptions with pan. 1.17 +func (s Subscription) GetSQLTableName() string { 1.18 + return "subscriptions" 1.19 +} 1.20 + 1.21 +func (p Postgres) resetSQL() *pan.Query { 1.22 + var sub Subscription 1.23 + query := pan.New(pan.POSTGRES, "TRUNCATE "+pan.GetTableName(sub)) 1.24 + return query.FlushExpressions(" ") 1.25 +} 1.26 + 1.27 +func (p Postgres) reset() error { 1.28 + query := p.resetSQL() 1.29 + _, err := p.Exec(query.String(), query.Args...) 1.30 + if err != nil { 1.31 + return err 1.32 + } 1.33 + return nil 1.34 +} 1.35 + 1.36 +func (p Postgres) createSubscriptionSQL(sub Subscription) *pan.Query { 1.37 + fields, values := pan.GetFields(sub) 1.38 + query := pan.New(pan.POSTGRES, "INSERT INTO "+pan.GetTableName(sub)) 1.39 + query.Include("(" + pan.QueryList(fields) + ")") 1.40 + query.Include("VALUES") 1.41 + query.Include("("+pan.VariableList(len(values))+")", values...) 1.42 + return query.FlushExpressions(" ") 1.43 +} 1.44 + 1.45 +func (p Postgres) createSubscription(sub Subscription) error { 1.46 + query := p.createSubscriptionSQL(sub) 1.47 + _, err := p.Exec(query.String(), query.Args...) 1.48 + if e, ok := err.(*pq.Error); ok && e.Constraint == "subscriptions_pkey" { 1.49 + err = ErrSubscriptionAlreadyExists 1.50 + } else if e, ok := err.(*pq.Error); ok && e.Constraint == "subscriptions_stripe_customer_key" { 1.51 + err = ErrStripeCustomerAlreadyExists 1.52 + } 1.53 + return err 1.54 +} 1.55 + 1.56 +func (p Postgres) updateSubscriptionSQL(id uuid.ID, change SubscriptionChange) *pan.Query { 1.57 + var sub Subscription 1.58 + query := pan.New(pan.POSTGRES, "UPDATE "+pan.GetTableName(sub)+" SET") 1.59 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "StripeCustomer")+" = ?", change.StripeCustomer) 1.60 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "Amount")+" = ?", change.Amount) 1.61 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "Period")+" = ?", change.Period) 1.62 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "BeginCharging")+" = ?", change.BeginCharging) 1.63 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "LastCharged")+" = ?", change.LastCharged) 1.64 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "LastNotified")+" = ?", change.LastNotified) 1.65 + query.IncludeIfNotNil(pan.GetUnquotedColumn(sub, "InLockout")+" = ?", change.InLockout) 1.66 + query.FlushExpressions(", ") 1.67 + query.IncludeWhere() 1.68 + query.Include(pan.GetUnquotedColumn(sub, "UserID")+" = ?", id) 1.69 + return query.FlushExpressions(" ") 1.70 +} 1.71 + 1.72 +func (p Postgres) updateSubscription(id uuid.ID, change SubscriptionChange) error { 1.73 + if change.IsEmpty() { 1.74 + return ErrSubscriptionChangeEmpty 1.75 + } 1.76 + 1.77 + query := p.updateSubscriptionSQL(id, change) 1.78 + res, err := p.Exec(query.String(), query.Args...) 1.79 + if e, ok := err.(*pq.Error); ok && e.Constraint == "subscriptions_stripe_customer_key" { 1.80 + return ErrStripeCustomerAlreadyExists 1.81 + } else if err != nil { 1.82 + return err 1.83 + } 1.84 + rows, err := res.RowsAffected() 1.85 + if err != nil { 1.86 + return err 1.87 + } 1.88 + if rows < 1 { 1.89 + return ErrSubscriptionNotFound 1.90 + } 1.91 + return nil 1.92 +} 1.93 + 1.94 +func (p Postgres) deleteSubscriptionSQL(id uuid.ID) *pan.Query { 1.95 + var sub Subscription 1.96 + query := pan.New(pan.POSTGRES, "DELETE FROM "+pan.GetTableName(sub)) 1.97 + query.IncludeWhere() 1.98 + query.Include(pan.GetUnquotedColumn(sub, "UserID")+" = ?", id) 1.99 + return query.FlushExpressions(" ") 1.100 +} 1.101 + 1.102 +func (p Postgres) deleteSubscription(id uuid.ID) error { 1.103 + query := p.deleteSubscriptionSQL(id) 1.104 + res, err := p.Exec(query.String(), query.Args...) 1.105 + if err != nil { 1.106 + return err 1.107 + } 1.108 + rows, err := res.RowsAffected() 1.109 + if err != nil { 1.110 + return err 1.111 + } 1.112 + if rows < 1 { 1.113 + return ErrSubscriptionNotFound 1.114 + } 1.115 + return nil 1.116 +} 1.117 + 1.118 +func (p Postgres) listSubscriptionsLastChargedBeforeSQL(cutoff time.Time) *pan.Query { 1.119 + var sub Subscription 1.120 + fields, _ := pan.GetFields(sub) 1.121 + query := pan.New(pan.POSTGRES, "SELECT "+pan.QueryList(fields)+" FROM "+pan.GetTableName(sub)) 1.122 + query.IncludeWhere() 1.123 + query.Include(pan.GetUnquotedColumn(sub, "LastCharged")+" < ?", cutoff) 1.124 + query.IncludeOrder(pan.GetUnquotedColumn(sub, "LastCharged") + " ASC") 1.125 + return query.FlushExpressions(" ") 1.126 +} 1.127 + 1.128 +func (p Postgres) listSubscriptionsLastChargedBefore(cutoff time.Time) ([]Subscription, error) { 1.129 + var results []Subscription 1.130 + query := p.listSubscriptionsLastChargedBeforeSQL(cutoff) 1.131 + rows, err := p.Query(query.String(), query.Args...) 1.132 + if err != nil { 1.133 + return results, err 1.134 + } 1.135 + for rows.Next() { 1.136 + var sub Subscription 1.137 + err := pan.Unmarshal(rows, &sub) 1.138 + if err != nil { 1.139 + return results, err 1.140 + } 1.141 + results = append(results, sub) 1.142 + } 1.143 + if err := rows.Err(); err != nil { 1.144 + return results, err 1.145 + } 1.146 + return results, nil 1.147 +} 1.148 + 1.149 +func (p Postgres) getSubscriptionsSQL(ids []uuid.ID) *pan.Query { 1.150 + var sub Subscription 1.151 + fields, _ := pan.GetFields(sub) 1.152 + intIDs := make([]interface{}, len(ids)) 1.153 + for pos, id := range ids { 1.154 + intIDs[pos] = id 1.155 + } 1.156 + query := pan.New(pan.POSTGRES, "SELECT "+pan.QueryList(fields)+" FROM "+pan.GetTableName(sub)) 1.157 + query.IncludeWhere() 1.158 + query.Include(pan.GetUnquotedColumn(sub, "UserID") + " IN") 1.159 + query.Include("("+pan.VariableList(len(intIDs))+")", intIDs...) 1.160 + return query.FlushExpressions(" ") 1.161 +} 1.162 + 1.163 +func (p Postgres) getSubscriptions(ids []uuid.ID) (map[string]Subscription, error) { 1.164 + results := map[string]Subscription{} 1.165 + if len(ids) < 1 { 1.166 + return results, ErrNoSubscriptionID 1.167 + } 1.168 + query := p.getSubscriptionsSQL(ids) 1.169 + rows, err := p.Query(query.String(), query.Args...) 1.170 + if err != nil { 1.171 + return results, err 1.172 + } 1.173 + for rows.Next() { 1.174 + var sub Subscription 1.175 + err := pan.Unmarshal(rows, &sub) 1.176 + if err != nil { 1.177 + return results, err 1.178 + } 1.179 + results[sub.UserID.String()] = sub 1.180 + } 1.181 + if err := rows.Err(); err != nil { 1.182 + return results, err 1.183 + } 1.184 + return results, nil 1.185 +} 1.186 + 1.187 +func (p Postgres) getSubscriptionStatsSQL() *pan.Query { 1.188 + var sub Subscription 1.189 + amountColumn := pan.GetUnquotedColumn(sub, "Amount") 1.190 + query := pan.New(pan.POSTGRES, "SELECT") 1.191 + query.Include("COUNT(*), SUM(" + amountColumn + "), AVG(" + amountColumn + ")") 1.192 + query.Include("FROM " + pan.GetTableName(sub)) 1.193 + return query.FlushExpressions(" ") 1.194 +} 1.195 + 1.196 +func (p Postgres) getSubscriptionStats() (SubscriptionStats, error) { 1.197 + query := p.getSubscriptionStatsSQL() 1.198 + rows, err := p.Query(query.String(), query.Args...) 1.199 + if err != nil { 1.200 + return SubscriptionStats{}, err 1.201 + } 1.202 + var stats SubscriptionStats 1.203 + for rows.Next() { 1.204 + var number, total sql.NullInt64 1.205 + var mean sql.NullFloat64 1.206 + if err := rows.Scan(number, total, mean); err != nil { 1.207 + return stats, err 1.208 + } 1.209 + if number.Valid { 1.210 + stats.Number = number.Int64 1.211 + } 1.212 + if total.Valid { 1.213 + stats.TotalAmount = total.Int64 1.214 + } 1.215 + if mean.Valid { 1.216 + stats.MeanAmount = mean.Float64 1.217 + } 1.218 + } 1.219 + if err := rows.Err(); err != nil { 1.220 + return stats, err 1.221 + } 1.222 + return stats, nil 1.223 +}