dex/storage/sql/crud.go

999 lines
25 KiB
Go
Raw Normal View History

package sql
import (
"database/sql"
"database/sql/driver"
"encoding/json"
"errors"
"fmt"
2016-10-06 04:34:11 +05:30
"strings"
"time"
"github.com/dexidp/dex/storage"
)
// TODO(ericchiang): The update, insert, and select methods queries are all
// very repetitive. Consider creating them programmatically.
// keysRowID is the ID of the only row we expect to populate the "keys" table.
const keysRowID = "keys"
// encoder wraps the underlying value in a JSON marshaler which is automatically
// called by the database/sql package.
//
// s := []string{"planes", "bears"}
// err := db.Exec(`insert into t1 (id, things) values (1, $1)`, encoder(s))
// if err != nil {
// // handle error
// }
//
// var r []byte
// err = db.QueryRow(`select things from t1 where id = 1;`).Scan(&r)
// if err != nil {
// // handle error
// }
// fmt.Printf("%s\n", r) // ["planes","bears"]
//
func encoder(i interface{}) driver.Valuer {
return jsonEncoder{i}
}
// decoder wraps the underlying value in a JSON unmarshaler which can then be passed
// to a database Scan() method.
func decoder(i interface{}) sql.Scanner {
return jsonDecoder{i}
}
type jsonEncoder struct {
i interface{}
}
func (j jsonEncoder) Value() (driver.Value, error) {
b, err := json.Marshal(j.i)
if err != nil {
return nil, fmt.Errorf("marshal: %v", err)
}
return b, nil
}
type jsonDecoder struct {
i interface{}
}
func (j jsonDecoder) Scan(dest interface{}) error {
if dest == nil {
return errors.New("nil value")
}
b, ok := dest.([]byte)
if !ok {
return fmt.Errorf("expected []byte got %T", dest)
}
if err := json.Unmarshal(b, &j.i); err != nil {
return fmt.Errorf("unmarshal: %v", err)
}
return nil
}
// Abstract conn vs trans.
type querier interface {
QueryRow(query string, args ...interface{}) *sql.Row
}
// Abstract row vs rows.
type scanner interface {
Scan(dest ...interface{}) error
}
func (c *conn) GarbageCollect(now time.Time) (storage.GCResult, error) {
result := storage.GCResult{}
r, err := c.Exec(`delete from auth_request where expiry < $1`, now)
if err != nil {
return result, fmt.Errorf("gc auth_request: %v", err)
}
if n, err := r.RowsAffected(); err == nil {
result.AuthRequests = n
}
r, err = c.Exec(`delete from auth_code where expiry < $1`, now)
if err != nil {
return result, fmt.Errorf("gc auth_code: %v", err)
}
if n, err := r.RowsAffected(); err == nil {
result.AuthCodes = n
}
r, err = c.Exec(`delete from device_request where expiry < $1`, now)
if err != nil {
return result, fmt.Errorf("gc device_request: %v", err)
}
if n, err := r.RowsAffected(); err == nil {
result.DeviceRequests = n
}
r, err = c.Exec(`delete from device_token where expiry < $1`, now)
if err != nil {
return result, fmt.Errorf("gc device_token: %v", err)
}
if n, err := r.RowsAffected(); err == nil {
result.DeviceTokens = n
}
return result, err
}
func (c *conn) CreateAuthRequest(a storage.AuthRequest) error {
_, err := c.Exec(`
insert into auth_request (
id, client_id, response_types, scopes, redirect_uri, nonce, state,
force_approval_prompt, logged_in,
2018-01-30 02:37:46 +05:30
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data,
expiry
)
values (
2019-04-18 19:33:27 +05:30
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18
);
`,
a.ID, a.ClientID, encoder(a.ResponseTypes), encoder(a.Scopes), a.RedirectURI, a.Nonce, a.State,
a.ForceApprovalPrompt, a.LoggedIn,
a.Claims.UserID, a.Claims.Username, a.Claims.PreferredUsername,
a.Claims.Email, a.Claims.EmailVerified, encoder(a.Claims.Groups),
2019-04-18 19:33:27 +05:30
a.ConnectorID, a.ConnectorData,
a.Expiry,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert auth request: %v", err)
}
return nil
}
func (c *conn) UpdateAuthRequest(id string, updater func(a storage.AuthRequest) (storage.AuthRequest, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
r, err := getAuthRequest(tx, id)
if err != nil {
return err
}
a, err := updater(r)
if err != nil {
return err
}
_, err = tx.Exec(`
update auth_request
set
client_id = $1, response_types = $2, scopes = $3, redirect_uri = $4,
nonce = $5, state = $6, force_approval_prompt = $7, logged_in = $8,
claims_user_id = $9, claims_username = $10, claims_preferred_username = $11,
claims_email = $12, claims_email_verified = $13,
claims_groups = $14,
2019-04-18 19:33:27 +05:30
connector_id = $15, connector_data = $16,
expiry = $17
where id = $18;
`,
a.ClientID, encoder(a.ResponseTypes), encoder(a.Scopes), a.RedirectURI, a.Nonce, a.State,
a.ForceApprovalPrompt, a.LoggedIn,
a.Claims.UserID, a.Claims.Username, a.Claims.PreferredUsername,
a.Claims.Email, a.Claims.EmailVerified,
encoder(a.Claims.Groups),
2019-04-18 19:33:27 +05:30
a.ConnectorID, a.ConnectorData,
2016-10-06 04:34:11 +05:30
a.Expiry, r.ID,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update auth request: %v", err)
}
return nil
})
}
func (c *conn) GetAuthRequest(id string) (storage.AuthRequest, error) {
2018-11-29 12:54:13 +05:30
return getAuthRequest(c, id)
}
func getAuthRequest(q querier, id string) (a storage.AuthRequest, err error) {
err = q.QueryRow(`
2018-01-30 02:37:46 +05:30
select
id, client_id, response_types, scopes, redirect_uri, nonce, state,
force_approval_prompt, logged_in,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data, expiry
from auth_request where id = $1;
`, id).Scan(
&a.ID, &a.ClientID, decoder(&a.ResponseTypes), decoder(&a.Scopes), &a.RedirectURI, &a.Nonce, &a.State,
&a.ForceApprovalPrompt, &a.LoggedIn,
&a.Claims.UserID, &a.Claims.Username, &a.Claims.PreferredUsername,
&a.Claims.Email, &a.Claims.EmailVerified,
decoder(&a.Claims.Groups),
2019-04-18 19:33:27 +05:30
&a.ConnectorID, &a.ConnectorData, &a.Expiry,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return a, storage.ErrNotFound
}
return a, fmt.Errorf("select auth request: %v", err)
}
return a, nil
}
func (c *conn) CreateAuthCode(a storage.AuthCode) error {
_, err := c.Exec(`
insert into auth_code (
id, client_id, scopes, nonce, redirect_uri,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data,
expiry
)
2019-04-18 19:33:27 +05:30
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14);
`,
a.ID, a.ClientID, encoder(a.Scopes), a.Nonce, a.RedirectURI, a.Claims.UserID,
a.Claims.Username, a.Claims.PreferredUsername, a.Claims.Email, a.Claims.EmailVerified,
2019-04-18 19:33:27 +05:30
encoder(a.Claims.Groups), a.ConnectorID, a.ConnectorData, a.Expiry,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert auth code: %v", err)
}
return nil
}
func (c *conn) GetAuthCode(id string) (a storage.AuthCode, err error) {
err = c.QueryRow(`
select
id, client_id, scopes, nonce, redirect_uri,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data,
expiry
from auth_code where id = $1;
`, id).Scan(
&a.ID, &a.ClientID, decoder(&a.Scopes), &a.Nonce, &a.RedirectURI, &a.Claims.UserID,
&a.Claims.Username, &a.Claims.PreferredUsername, &a.Claims.Email, &a.Claims.EmailVerified,
2019-04-18 19:33:27 +05:30
decoder(&a.Claims.Groups), &a.ConnectorID, &a.ConnectorData, &a.Expiry,
)
if err != nil {
if err == sql.ErrNoRows {
return a, storage.ErrNotFound
}
return a, fmt.Errorf("select auth code: %v", err)
}
return a, nil
}
func (c *conn) CreateRefresh(r storage.RefreshToken) error {
_, err := c.Exec(`
insert into refresh_token (
id, client_id, scopes, nonce,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data,
token, created_at, last_used
)
2019-04-18 19:33:27 +05:30
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15);
`,
r.ID, r.ClientID, encoder(r.Scopes), r.Nonce,
r.Claims.UserID, r.Claims.Username, r.Claims.PreferredUsername,
r.Claims.Email, r.Claims.EmailVerified,
encoder(r.Claims.Groups),
2019-04-18 19:33:27 +05:30
r.ConnectorID, r.ConnectorData,
r.Token, r.CreatedAt, r.LastUsed,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
2018-11-29 12:54:13 +05:30
return fmt.Errorf("insert refresh_token: %v", err)
}
return nil
}
func (c *conn) UpdateRefreshToken(id string, updater func(old storage.RefreshToken) (storage.RefreshToken, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
r, err := getRefresh(tx, id)
if err != nil {
return err
}
if r, err = updater(r); err != nil {
return err
}
_, err = tx.Exec(`
update refresh_token
set
client_id = $1,
scopes = $2,
nonce = $3,
claims_user_id = $4,
claims_username = $5,
claims_preferred_username = $6,
claims_email = $7,
claims_email_verified = $8,
claims_groups = $9,
connector_id = $10,
2019-04-18 19:33:27 +05:30
connector_data = $11,
token = $12,
created_at = $13,
last_used = $14
where
2019-04-18 19:33:27 +05:30
id = $15
`,
r.ClientID, encoder(r.Scopes), r.Nonce,
r.Claims.UserID, r.Claims.Username, r.Claims.PreferredUsername,
r.Claims.Email, r.Claims.EmailVerified,
encoder(r.Claims.Groups),
2019-04-18 19:33:27 +05:30
r.ConnectorID, r.ConnectorData,
r.Token, r.CreatedAt, r.LastUsed, id,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update refresh token: %v", err)
}
return nil
})
}
func (c *conn) GetRefresh(id string) (storage.RefreshToken, error) {
2018-11-29 12:54:13 +05:30
return getRefresh(c, id)
}
func getRefresh(q querier, id string) (storage.RefreshToken, error) {
return scanRefresh(q.QueryRow(`
select
id, client_id, scopes, nonce,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified,
claims_groups,
2019-04-18 19:33:27 +05:30
connector_id, connector_data,
token, created_at, last_used
from refresh_token where id = $1;
`, id))
}
func (c *conn) ListRefreshTokens() ([]storage.RefreshToken, error) {
rows, err := c.Query(`
select
id, client_id, scopes, nonce,
claims_user_id, claims_username, claims_preferred_username,
claims_email, claims_email_verified, claims_groups,
connector_id, connector_data,
token, created_at, last_used
from refresh_token;
`)
if err != nil {
2018-11-29 12:54:13 +05:30
return nil, fmt.Errorf("query: %v", err)
}
var tokens []storage.RefreshToken
for rows.Next() {
r, err := scanRefresh(rows)
if err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
tokens = append(tokens, r)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("scan: %v", err)
}
return tokens, nil
}
func scanRefresh(s scanner) (r storage.RefreshToken, err error) {
err = s.Scan(
&r.ID, &r.ClientID, decoder(&r.Scopes), &r.Nonce,
&r.Claims.UserID, &r.Claims.Username, &r.Claims.PreferredUsername,
&r.Claims.Email, &r.Claims.EmailVerified,
decoder(&r.Claims.Groups),
2019-04-18 19:33:27 +05:30
&r.ConnectorID, &r.ConnectorData,
&r.Token, &r.CreatedAt, &r.LastUsed,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return r, storage.ErrNotFound
}
return r, fmt.Errorf("scan refresh_token: %v", err)
}
return r, nil
}
func (c *conn) UpdateKeys(updater func(old storage.Keys) (storage.Keys, error)) error {
return c.ExecTx(func(tx *trans) error {
firstUpdate := false
// TODO(ericchiang): errors may cause a transaction be rolled back by the SQL
// server. Test this, and consider adding a COUNT() command beforehand.
old, err := getKeys(tx)
2018-11-29 12:54:13 +05:30
if err != nil {
if err != storage.ErrNotFound {
return fmt.Errorf("get keys: %v", err)
}
firstUpdate = true
old = storage.Keys{}
}
nk, err := updater(old)
if err != nil {
return err
}
if firstUpdate {
_, err = tx.Exec(`
insert into keys (
id, verification_keys, signing_key, signing_key_pub, next_rotation
)
values ($1, $2, $3, $4, $5);
`,
keysRowID, encoder(nk.VerificationKeys), encoder(nk.SigningKey),
encoder(nk.SigningKeyPub), nk.NextRotation,
)
if err != nil {
2018-11-29 12:54:13 +05:30
return fmt.Errorf("insert: %v", err)
}
} else {
_, err = tx.Exec(`
update keys
2018-01-30 02:37:46 +05:30
set
verification_keys = $1,
signing_key = $2,
2016-10-15 00:26:50 +05:30
signing_key_pub = $3,
next_rotation = $4
where id = $5;
`,
encoder(nk.VerificationKeys), encoder(nk.SigningKey),
encoder(nk.SigningKeyPub), nk.NextRotation, keysRowID,
)
if err != nil {
2018-11-29 12:54:13 +05:30
return fmt.Errorf("update: %v", err)
}
}
return nil
})
}
2018-11-29 12:54:13 +05:30
func (c *conn) GetKeys() (keys storage.Keys, err error) {
return getKeys(c)
}
func getKeys(q querier) (keys storage.Keys, err error) {
err = q.QueryRow(`
select
verification_keys, signing_key, signing_key_pub, next_rotation
from keys
2016-10-15 00:26:50 +05:30
where id=$1
`, keysRowID).Scan(
decoder(&keys.VerificationKeys), decoder(&keys.SigningKey),
decoder(&keys.SigningKeyPub), &keys.NextRotation,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return keys, storage.ErrNotFound
}
return keys, fmt.Errorf("query keys: %v", err)
}
return keys, nil
}
func (c *conn) UpdateClient(id string, updater func(old storage.Client) (storage.Client, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
cli, err := getClient(tx, id)
if err != nil {
return err
}
nc, err := updater(cli)
if err != nil {
return err
}
_, err = tx.Exec(`
update client
set
secret = $1,
redirect_uris = $2,
trusted_peers = $3,
public = $4,
name = $5,
logo_url = $6
where id = $7;
`, nc.Secret, encoder(nc.RedirectURIs), encoder(nc.TrustedPeers), nc.Public, nc.Name, nc.LogoURL, id,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update client: %v", err)
}
return nil
})
}
func (c *conn) CreateClient(cli storage.Client) error {
_, err := c.Exec(`
insert into client (
id, secret, redirect_uris, trusted_peers, public, name, logo_url
)
values ($1, $2, $3, $4, $5, $6, $7);
`,
cli.ID, cli.Secret, encoder(cli.RedirectURIs), encoder(cli.TrustedPeers),
cli.Public, cli.Name, cli.LogoURL,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert client: %v", err)
}
return nil
}
func getClient(q querier, id string) (storage.Client, error) {
return scanClient(q.QueryRow(`
select
id, secret, redirect_uris, trusted_peers, public, name, logo_url
from client where id = $1;
`, id))
}
func (c *conn) GetClient(id string) (storage.Client, error) {
2018-11-29 12:54:13 +05:30
return getClient(c, id)
}
func (c *conn) ListClients() ([]storage.Client, error) {
rows, err := c.Query(`
select
id, secret, redirect_uris, trusted_peers, public, name, logo_url
from client;
`)
if err != nil {
return nil, err
}
var clients []storage.Client
for rows.Next() {
cli, err := scanClient(rows)
if err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
clients = append(clients, cli)
}
if err := rows.Err(); err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
return clients, nil
}
func scanClient(s scanner) (cli storage.Client, err error) {
err = s.Scan(
&cli.ID, &cli.Secret, decoder(&cli.RedirectURIs), decoder(&cli.TrustedPeers),
&cli.Public, &cli.Name, &cli.LogoURL,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return cli, storage.ErrNotFound
}
return cli, fmt.Errorf("get client: %v", err)
}
return cli, nil
}
2016-10-06 04:34:11 +05:30
func (c *conn) CreatePassword(p storage.Password) error {
p.Email = strings.ToLower(p.Email)
_, err := c.Exec(`
insert into password (
email, hash, username, user_id
)
values (
$1, $2, $3, $4
);
`,
p.Email, p.Hash, p.Username, p.UserID,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
2016-10-06 04:34:11 +05:30
return fmt.Errorf("insert password: %v", err)
}
return nil
}
func (c *conn) UpdatePassword(email string, updater func(p storage.Password) (storage.Password, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
2016-10-06 04:34:11 +05:30
p, err := getPassword(tx, email)
if err != nil {
return err
}
np, err := updater(p)
if err != nil {
return err
}
_, err = tx.Exec(`
update password
set
hash = $1, username = $2, user_id = $3
where email = $4;
`,
np.Hash, np.Username, np.UserID, p.Email,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update password: %v", err)
}
return nil
2016-10-06 04:34:11 +05:30
})
}
func (c *conn) GetPassword(email string) (storage.Password, error) {
2018-11-29 12:54:13 +05:30
return getPassword(c, email)
2016-10-06 04:34:11 +05:30
}
func getPassword(q querier, email string) (p storage.Password, err error) {
return scanPassword(q.QueryRow(`
2016-10-06 04:34:11 +05:30
select
email, hash, username, user_id
from password where email = $1;
`, strings.ToLower(email)))
}
func (c *conn) ListPasswords() ([]storage.Password, error) {
rows, err := c.Query(`
select
email, hash, username, user_id
from password;
`)
if err != nil {
return nil, err
}
var passwords []storage.Password
for rows.Next() {
p, err := scanPassword(rows)
if err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
passwords = append(passwords, p)
}
if err := rows.Err(); err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
return passwords, nil
}
func scanPassword(s scanner) (p storage.Password, err error) {
err = s.Scan(
2016-10-06 04:34:11 +05:30
&p.Email, &p.Hash, &p.Username, &p.UserID,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return p, storage.ErrNotFound
}
return p, fmt.Errorf("select password: %v", err)
2016-10-06 04:34:11 +05:30
}
return p, nil
}
func (c *conn) CreateOfflineSessions(s storage.OfflineSessions) error {
_, err := c.Exec(`
insert into offline_session (
2018-01-30 02:37:46 +05:30
user_id, conn_id, refresh, connector_data
)
values (
2018-01-30 02:37:46 +05:30
$1, $2, $3, $4
);
`,
2018-01-30 02:37:46 +05:30
s.UserID, s.ConnID, encoder(s.Refresh), s.ConnectorData,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert offline session: %v", err)
}
return nil
}
func (c *conn) UpdateOfflineSessions(userID string, connID string, updater func(s storage.OfflineSessions) (storage.OfflineSessions, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
s, err := getOfflineSessions(tx, userID, connID)
if err != nil {
return err
}
newSession, err := updater(s)
if err != nil {
return err
}
_, err = tx.Exec(`
update offline_session
set
2018-01-30 16:49:08 +05:30
refresh = $1,
2018-01-30 02:37:46 +05:30
connector_data = $2
where user_id = $3 AND conn_id = $4;
`,
2018-02-15 16:30:06 +05:30
encoder(newSession.Refresh), newSession.ConnectorData, s.UserID, s.ConnID,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update offline session: %v", err)
}
return nil
})
}
func (c *conn) GetOfflineSessions(userID string, connID string) (storage.OfflineSessions, error) {
2018-11-29 12:54:13 +05:30
return getOfflineSessions(c, userID, connID)
}
func getOfflineSessions(q querier, userID string, connID string) (storage.OfflineSessions, error) {
return scanOfflineSessions(q.QueryRow(`
select
2018-01-30 02:37:46 +05:30
user_id, conn_id, refresh, connector_data
from offline_session
where user_id = $1 AND conn_id = $2;
`, userID, connID))
}
func scanOfflineSessions(s scanner) (o storage.OfflineSessions, err error) {
err = s.Scan(
2018-01-30 02:37:46 +05:30
&o.UserID, &o.ConnID, decoder(&o.Refresh), &o.ConnectorData,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return o, storage.ErrNotFound
}
return o, fmt.Errorf("select offline session: %v", err)
}
return o, nil
}
func (c *conn) CreateConnector(connector storage.Connector) error {
_, err := c.Exec(`
insert into connector (
id, type, name, resource_version, config
)
values (
$1, $2, $3, $4, $5
);
`,
connector.ID, connector.Type, connector.Name, connector.ResourceVersion, connector.Config,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert connector: %v", err)
}
return nil
}
func (c *conn) UpdateConnector(id string, updater func(s storage.Connector) (storage.Connector, error)) error {
2018-11-29 12:54:13 +05:30
return c.ExecTx(func(tx *trans) error {
connector, err := getConnector(tx, id)
if err != nil {
return err
}
newConn, err := updater(connector)
if err != nil {
return err
}
_, err = tx.Exec(`
update connector
2018-01-30 02:37:46 +05:30
set
type = $1,
name = $2,
resource_version = $3,
config = $4
where id = $5;
`,
newConn.Type, newConn.Name, newConn.ResourceVersion, newConn.Config, connector.ID,
)
2018-11-29 12:54:13 +05:30
if err != nil {
return fmt.Errorf("update connector: %v", err)
}
return nil
})
}
func (c *conn) GetConnector(id string) (storage.Connector, error) {
2018-11-29 12:54:13 +05:30
return getConnector(c, id)
}
func getConnector(q querier, id string) (storage.Connector, error) {
return scanConnector(q.QueryRow(`
select
id, type, name, resource_version, config
from connector
where id = $1;
`, id))
}
func scanConnector(s scanner) (c storage.Connector, err error) {
err = s.Scan(
&c.ID, &c.Type, &c.Name, &c.ResourceVersion, &c.Config,
)
if err != nil {
2018-11-29 12:54:13 +05:30
if err == sql.ErrNoRows {
return c, storage.ErrNotFound
}
return c, fmt.Errorf("select connector: %v", err)
}
return c, nil
}
func (c *conn) ListConnectors() ([]storage.Connector, error) {
rows, err := c.Query(`
select
id, type, name, resource_version, config
from connector;
`)
if err != nil {
return nil, err
}
var connectors []storage.Connector
for rows.Next() {
conn, err := scanConnector(rows)
if err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
connectors = append(connectors, conn)
}
if err := rows.Err(); err != nil {
2018-11-29 12:54:13 +05:30
return nil, err
}
return connectors, nil
}
2016-10-06 04:34:11 +05:30
func (c *conn) DeleteAuthRequest(id string) error { return c.delete("auth_request", "id", id) }
func (c *conn) DeleteAuthCode(id string) error { return c.delete("auth_code", "id", id) }
func (c *conn) DeleteClient(id string) error { return c.delete("client", "id", id) }
func (c *conn) DeleteRefresh(id string) error { return c.delete("refresh_token", "id", id) }
func (c *conn) DeletePassword(email string) error {
return c.delete("password", "email", strings.ToLower(email))
}
func (c *conn) DeleteConnector(id string) error { return c.delete("connector", "id", id) }
func (c *conn) DeleteOfflineSessions(userID string, connID string) error {
result, err := c.Exec(`delete from offline_session where user_id = $1 AND conn_id = $2`, userID, connID)
if err != nil {
return fmt.Errorf("delete offline_session: user_id = %s, conn_id = %s", userID, connID)
}
// For now mandate that the driver implements RowsAffected. If we ever need to support
// a driver that doesn't implement this, we can run this in a transaction with a get beforehand.
n, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("rows affected: %v", err)
}
if n < 1 {
return storage.ErrNotFound
}
return nil
}
// Do NOT call directly. Does not escape table.
2016-10-06 04:34:11 +05:30
func (c *conn) delete(table, field, id string) error {
result, err := c.Exec(`delete from `+table+` where `+field+` = $1`, id)
if err != nil {
return fmt.Errorf("delete %s: %v", table, id)
}
// For now mandate that the driver implements RowsAffected. If we ever need to support
// a driver that doesn't implement this, we can run this in a transaction with a get beforehand.
n, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("rows affected: %v", err)
}
if n < 1 {
return storage.ErrNotFound
}
return nil
}
func (c *conn) CreateDeviceRequest(d storage.DeviceRequest) error {
_, err := c.Exec(`
insert into device_request (
user_code, device_code, client_id, client_secret, scopes, expiry
)
values (
$1, $2, $3, $4, $5, $6
);`,
d.UserCode, d.DeviceCode, d.ClientID, d.ClientSecret, encoder(d.Scopes), d.Expiry,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert device request: %v", err)
}
return nil
}
func (c *conn) CreateDeviceToken(t storage.DeviceToken) error {
_, err := c.Exec(`
insert into device_token (
device_code, status, token, expiry, last_request, poll_interval
)
values (
$1, $2, $3, $4, $5, $6
);`,
t.DeviceCode, t.Status, t.Token, t.Expiry, t.LastRequestTime, t.PollIntervalSeconds,
)
if err != nil {
if c.alreadyExistsCheck(err) {
return storage.ErrAlreadyExists
}
return fmt.Errorf("insert device token: %v", err)
}
return nil
}
func (c *conn) GetDeviceRequest(userCode string) (storage.DeviceRequest, error) {
return getDeviceRequest(c, userCode)
}
func getDeviceRequest(q querier, userCode string) (d storage.DeviceRequest, err error) {
err = q.QueryRow(`
select
device_code, client_id, client_secret, scopes, expiry
from device_request where user_code = $1;
`, userCode).Scan(
&d.DeviceCode, &d.ClientID, &d.ClientSecret, decoder(&d.Scopes), &d.Expiry,
)
if err != nil {
if err == sql.ErrNoRows {
return d, storage.ErrNotFound
}
return d, fmt.Errorf("select device token: %v", err)
}
d.UserCode = userCode
return d, nil
}
func (c *conn) GetDeviceToken(deviceCode string) (storage.DeviceToken, error) {
return getDeviceToken(c, deviceCode)
}
func getDeviceToken(q querier, deviceCode string) (a storage.DeviceToken, err error) {
err = q.QueryRow(`
select
status, token, expiry, last_request, poll_interval
from device_token where device_code = $1;
`, deviceCode).Scan(
&a.Status, &a.Token, &a.Expiry, &a.LastRequestTime, &a.PollIntervalSeconds,
)
if err != nil {
if err == sql.ErrNoRows {
return a, storage.ErrNotFound
}
return a, fmt.Errorf("select device token: %v", err)
}
a.DeviceCode = deviceCode
return a, nil
}
func (c *conn) UpdateDeviceToken(deviceCode string, updater func(old storage.DeviceToken) (storage.DeviceToken, error)) error {
return c.ExecTx(func(tx *trans) error {
r, err := getDeviceToken(tx, deviceCode)
if err != nil {
return err
}
if r, err = updater(r); err != nil {
return err
}
_, err = tx.Exec(`
update device_token
set
status = $1,
token = $2,
last_request = $3,
poll_interval = $4
where
device_code = $5
`,
r.Status, r.Token, r.LastRequestTime, r.PollIntervalSeconds, r.DeviceCode,
)
if err != nil {
return fmt.Errorf("update device token: %v", err)
}
return nil
})
}