24770254403

Committed 22 Apr 2026 09:13AM UTC coverage: 83.056% (+0.02%) from 83.041%

Build # 24770254403

Build Type

push

github

Committed by

web-flow

Commit Message

FSC v0.10.1 (#1577)

Signed-off-by: Angelo De Caro <adc@zurich.ibm.com>

Coverage Stats

33910 of 40828 relevant lines covered (83.06%)

17.08 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.0

/token/services/storage/db/sql/common/identity.go

/*
Copyright IBM Corp. All Rights Reserved.

SPDX-License-Identifier: Apache-2.0
*/

package common

import (
        "context"
        "database/sql"
        "fmt"

        "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
        "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/cache/secondcache"
        "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators"
        driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver"
        common2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/common"
        "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/common"
        q "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query"
        common3 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query/common"
        "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query/cond"
        "github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
        "github.com/hyperledger-labs/fabric-token-sdk/token"
        tdriver "github.com/hyperledger-labs/fabric-token-sdk/token/driver"
        idriver "github.com/hyperledger-labs/fabric-token-sdk/token/services/identity/driver"
        "github.com/hyperledger-labs/fabric-token-sdk/token/services/logging"
        "github.com/hyperledger-labs/fabric-token-sdk/token/services/storage"
        "github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/db/driver"
        "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils"
        cache2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/cache"
)

type cache[T any] interface {
        Get(key string) (T, bool)
        GetOrLoad(key string, loader func() (T, error)) (T, bool, error)
        Add(key string, value T)
        Delete(key string)
}

type dbTransaction interface {
        ExecContext(ctx context.Context, query string, args ...common3.Param) (sql.Result, error)
}

type identityTables struct {
        IdentityConfigurations string
        IdentityInfo           string
        Signers                string
}

// IdentityStore is a SQL-backed implementation of the IdentityStore interface.
type IdentityStore struct {
        readDB   *sql.DB
        writeDB  *sql.DB
        table    identityTables
        ci       common3.CondInterpreter
        notifier idriver.IdentityConfigurationNotifier

        signerInfoCache cache[bool]
        auditInfoCache  cache[[]byte]
        errorWrapper    driver2.SQLErrorWrapper
}

func newIdentityStore(
        readDB, writeDB *sql.DB,
        tables identityTables,
        singerInfoCache cache[bool],
        auditInfoCache cache[[]byte],
        ci common3.CondInterpreter,
        errorWrapper driver2.SQLErrorWrapper,
        notifier idriver.IdentityConfigurationNotifier,
) *IdentityStore {
        return &IdentityStore{
                readDB:          readDB,
                writeDB:         writeDB,
                table:           tables,
                signerInfoCache: singerInfoCache,
                auditInfoCache:  auditInfoCache,
                ci:              ci,
                errorWrapper:    errorWrapper,
                notifier:        notifier,
        }
}

func NewCachedIdentityStore(
        readDB, writeDB *sql.DB,
        tables TableNames,
        ci common3.CondInterpreter,
        errorWrapper driver2.SQLErrorWrapper,
) (*IdentityStore, error) {
        return NewIdentityStore(
                readDB,
                writeDB,
                tables,
                secondcache.NewTyped[bool](5000),
                secondcache.NewTyped[[]byte](5000),
                ci,
                errorWrapper,
        )
}

func NewNoCacheIdentityStore(
        readDB, writeDB *sql.DB,
        tables TableNames,
        ci common3.CondInterpreter,
        errorWrapper driver2.SQLErrorWrapper,
) (*IdentityStore, error) {
        return NewIdentityStore(
                readDB,
                writeDB,
                tables,
                cache2.NewNoCache[bool](),
                cache2.NewNoCache[[]byte](),
                ci,
                errorWrapper,
        )
}

func NewIdentityStore(
        readDB, writeDB *sql.DB,
        tables TableNames,
        signerInfoCache cache[bool],
        auditInfoCache cache[[]byte],
        ci common3.CondInterpreter,
        errorWrapper driver2.SQLErrorWrapper,
) (*IdentityStore, error) {
        return newIdentityStore(
                readDB,
                writeDB,
                identityTables{
                        IdentityConfigurations: tables.IdentityConfigurations,
                        IdentityInfo:           tables.IdentityInfo,
                        Signers:                tables.Signers,
                },
                signerInfoCache,
                auditInfoCache,
                ci,
                errorWrapper,
                nil,
        ), nil
}

// NewIdentityStoreWithNotifier creates a new IdentityStore with a notifier.
func NewIdentityStoreWithNotifier(
        readDB, writeDB *sql.DB,
        tables TableNames,
        signerInfoCache cache[bool],
        auditInfoCache cache[[]byte],
        ci common3.CondInterpreter,
        errorWrapper driver2.SQLErrorWrapper,
        notifier idriver.IdentityConfigurationNotifier,
) (*IdentityStore, error) {
        return newIdentityStore(
                readDB,
                writeDB,
                identityTables{
                        IdentityConfigurations: tables.IdentityConfigurations,
                        IdentityInfo:           tables.IdentityInfo,
                        Signers:                tables.Signers,
                },
                signerInfoCache,
                auditInfoCache,
                ci,
                errorWrapper,
                notifier,
        ), nil
}

func (db *IdentityStore) CreateSchema() error {
        return common.InitSchema(db.writeDB, []string{db.GetSchema()}...)
}

// AddConfiguration stores an identity configuration in the database.
// It also enqueues an event to the notifier if available.
func (db *IdentityStore) AddConfiguration(ctx context.Context, wp driver.IdentityConfiguration) error {
        query, args := q.InsertInto(db.table.IdentityConfigurations).
                Fields("id", "type", "url", "conf", "raw").
                Row(wp.ID, wp.Type, wp.URL, wp.Config, wp.Raw).
                Format()
        logging.Debug(logger, query, args)

        _, err := db.writeDB.ExecContext(ctx, query, args...)
        if err != nil {
                return err
        }

        return nil
}

// GetConfiguration returns the configuration with the given id, type, and url.
func (db *IdentityStore) GetConfiguration(ctx context.Context, id, typ, url string) (*driver.IdentityConfiguration, error) {
        query, args := q.Select().
                FieldsByName("id", "type", "url", "conf", "raw").
                From(q.Table(db.table.IdentityConfigurations)).
                Where(cond.And(cond.Eq("id", id), cond.Eq("type", typ), cond.Eq("url", url))).
                Format(db.ci)
        logging.Debug(logger, query, args)
        row := db.readDB.QueryRowContext(ctx, query, args...)
        c := &driver.IdentityConfiguration{}
        err := row.Scan(&c.ID, &c.Type, &c.URL, &c.Config, &c.Raw)
        if err != nil {
                if errors.Is(err, sql.ErrNoRows) {
                        return nil, nil
                }

                return nil, err
        }

        return c, nil
}

// IteratorConfigurations returns an iterator to all configurations of the given type.
func (db *IdentityStore) IteratorConfigurations(ctx context.Context, configurationType string) (idriver.IdentityConfigurationIterator, error) {
        query, args := q.Select().
                FieldsByName("id", "url", "conf", "raw").
                From(q.Table(db.table.IdentityConfigurations)).
                Where(cond.Eq("type", configurationType)).
                Format(db.ci)
        logging.Debug(logger, query, args)
        rows, err := db.readDB.QueryContext(ctx, query, args...)
        if err != nil {
                return nil, err
        }

        return common.NewIterator(rows, func(c *driver.IdentityConfiguration) error {
                c.Type = configurationType

                return rows.Scan(&c.ID, &c.URL, &c.Config, &c.Raw)
        }), nil
}

func (db *IdentityStore) ConfigurationExists(ctx context.Context, id, typ, url string) (bool, error) {
        query, args := q.Select().
                FieldsByName("id").
                From(q.Table(db.table.IdentityConfigurations)).
                Where(cond.And(cond.Eq("id", id), cond.Eq("type", typ), cond.Eq("url", url))).
                Format(db.ci)
        result, err := common.QueryUnique[string](db.readDB, query, args...)
        if err != nil {
                return false, errors.Wrapf(err, "failed getting configuration for [%s:%s:%s]", id, typ, url)
        }
        logger.DebugfContext(ctx, "found configuration for [%s:%s:%s]", id, typ, url)

        return len(result) != 0, nil
}

// Notifier returns the IdentityNotifier associated with this store.
func (db *IdentityStore) Notifier() (idriver.IdentityConfigurationNotifier, error) {
        if db.notifier == nil {
                return nil, storage.ErrNotSupported
        }

        return db.notifier, nil
}

func (db *IdentityStore) StoreIdentityData(ctx context.Context, id []byte, identityAudit []byte, tokenMetadata []byte, tokenMetadataAudit []byte) error {
        return db.storeIdentityData(ctx, db.writeDB, tdriver.Identity(id).UniqueID(), id, identityAudit, tokenMetadata, tokenMetadataAudit, true)
}

func (db *IdentityStore) GetAuditInfo(ctx context.Context, id []byte) ([]byte, error) {
        h := token.Identity(id).String()
        logger.DebugfContext(ctx, "get audit info for [%s]", h)

        value, _, err := db.auditInfoCache.GetOrLoad(h, func() ([]byte, error) {
                logger.DebugfContext(ctx, "load from backend identity data for [%s]", view.Identity(id))
                query, args := q.Select().
                        FieldsByName("identity_audit_info").
                        From(q.Table(db.table.IdentityInfo)).
                        Where(cond.Eq("identity_hash", h)).
                        Format(db.ci)

                return common.QueryUnique[[]byte](db.readDB, query, args...)
        })

        return value, err
}

func (db *IdentityStore) GetTokenInfo(ctx context.Context, id []byte) ([]byte, []byte, error) {
        h := token.Identity(id).String()
        logger.DebugfContext(ctx, "get identity data for [%s]", h)

        query, args := q.Select().
                FieldsByName("token_metadata", "token_metadata_audit_info").
                From(q.Table(db.table.IdentityInfo)).
                Where(cond.Eq("identity_hash", h)).
                Format(db.ci)
        logging.Debug(logger, query, args)

        row := db.readDB.QueryRowContext(ctx, query, args...)
        var tokenMetadata []byte
        var tokenMetadataAuditInfo []byte
        err := row.Scan(&tokenMetadata, &tokenMetadataAuditInfo)
        if err != nil {
                if errors.Is(err, sql.ErrNoRows) {
                        return nil, nil, nil
                }

                return nil, nil, errors.Wrapf(err, "error querying db")
        }

        return tokenMetadata, tokenMetadataAuditInfo, nil
}

func (db *IdentityStore) StoreSignerInfo(ctx context.Context, id tdriver.Identity, info []byte) error {
        _, err := db.storeSignerInfo(ctx, db.writeDB, id.UniqueID(), id, info, true)

        return err
}

func (db *IdentityStore) GetExistingSignerInfo(ctx context.Context, ids ...tdriver.Identity) ([]string, error) {
        idHashes := make([]string, len(ids))
        for i, id := range ids {
                idHashes[i] = id.UniqueID()
        }

        result := make([]string, 0)
        notFound := make([]string, 0)

        for _, idHash := range idHashes {
                if v, ok := db.signerInfoCache.Get(idHash); !ok {
                        notFound = append(notFound, idHash)
                } else if v {
                        result = append(result, idHash)
                }
        }
        if len(notFound) == 0 {
                return result, nil
        }

        idHashes = notFound
        notFound = make([]string, 0)
        for _, idHash := range idHashes {
                if v, ok := db.signerInfoCache.Get(idHash); !ok {
                        notFound = append(notFound, idHash)
                } else if v {
                        result = append(result, idHash)
                }
        }
        if len(notFound) == 0 {
                return result, nil
        }

        idHashes = notFound

        query, args := q.Select().
                FieldsByName("identity_hash").
                From(q.Table(db.table.Signers)).
                Where(cond.In("identity_hash", idHashes...)).
                Format(db.ci)

        logging.Debug(logger, query, args)
        rows, err := db.readDB.QueryContext(ctx, query, args...)
        if err != nil {
                return nil, errors.Wrapf(err, "error querying db")
        }
        it := common.NewIterator(rows, func(idHash *string) error { return rows.Scan(idHash) })

        found, err := iterators.Reduce(it, iterators.ToSet[string]())
        if err != nil {
                return nil, err
        }
        for _, idHash := range idHashes {
                db.signerInfoCache.Add(idHash, found.Contains(idHash))
        }

        return append(result, found.ToSlice()...), nil
}

func (db *IdentityStore) SignerInfoExists(ctx context.Context, id []byte) (bool, error) {
        existing, err := db.GetExistingSignerInfo(ctx, id)
        if err != nil {
                return false, err
        }

        return len(existing) > 0, nil
}

func (db *IdentityStore) GetSignerInfo(ctx context.Context, identity []byte) ([]byte, error) {
        query, args := q.Select().
                FieldsByName("info").
                From(q.Table(db.table.Signers)).
                Where(cond.Eq("identity_hash", token.Identity(identity).UniqueID())).
                Format(db.ci)

        return common.QueryUniqueContext[[]byte](ctx, db.readDB, query, args...)
}

func (db *IdentityStore) RegisterIdentityDescriptor(ctx context.Context, descriptor *idriver.IdentityDescriptor, alias tdriver.Identity) error {
        // store
        logger.DebugfContext(ctx, "register identity descriptor...")
        if err := db.registerIdentityDescriptor(ctx, descriptor, alias); err != nil {
                logger.ErrorfContext(ctx, "register identity descriptor...failed: %v", err)

                return err
        }
        logger.DebugfContext(ctx, "register identity descriptor...done")

        // update all caches
        logger.DebugfContext(ctx, "register identity descriptor...update caches...")
        h := descriptor.Identity.UniqueID()
        db.signerInfoCache.Add(h, true)
        if len(descriptor.AuditInfo) != 0 {
                db.auditInfoCache.Add(h, descriptor.AuditInfo)
        }
        if !alias.IsNone() && !descriptor.Identity.Equal(alias) {
                h = alias.UniqueID()
                db.signerInfoCache.Add(h, true)
                if len(descriptor.AuditInfo) != 0 {
                        db.auditInfoCache.Add(h, descriptor.AuditInfo)
                }
        }
        logger.DebugfContext(ctx, "register identity descriptor...update caches...done")

        return nil
}

func (db *IdentityStore) registerIdentityDescriptor(
        ctx context.Context,
        descriptor *idriver.IdentityDescriptor,
        alias tdriver.Identity,
) error {
        if descriptor == nil {
                return errors.New("identity descriptor is nil")
        }
        tx, err := db.writeDB.BeginTx(ctx, nil)
        if err != nil {
                return err
        }
        defer func() {
                if tx != nil {
                        if err := tx.Rollback(); err != nil {
                                logger.ErrorfContext(ctx, "failed closing connection: %s", err)
                        }
                }
        }()

        h := descriptor.Identity.UniqueID()

        exists, err := db.storeSignerInfo(ctx, tx, h, descriptor.Identity, descriptor.SignerInfo, false)
        if err != nil {
                return errors.Wrapf(err, "failed to store signer info for descriptor's identity")
        }
        if exists {
                // no need to continue
                return nil
        }

        if len(descriptor.AuditInfo) != 0 {
                err = db.storeIdentityData(ctx, tx, h, descriptor.Identity, descriptor.AuditInfo, nil, nil, false)
                if err != nil {
                        return errors.Wrapf(err, "failed to store audit info for descriptor's identity")
                }
        }

        if !alias.IsNone() && !descriptor.Identity.Equal(alias) {
                h = alias.UniqueID()
                _, err = db.storeSignerInfo(ctx, tx, h, alias, descriptor.SignerInfo, false)
                if err != nil {
                        return errors.Wrapf(err, "failed to store signer info for alias")
                }
                if len(descriptor.AuditInfo) != 0 {
                        err = db.storeIdentityData(ctx, tx, h, alias, descriptor.AuditInfo, nil, nil, false)
                        if err != nil {
                                return errors.Wrapf(err, "failed to store audit info for alias")
                        }
                }
        }
        if err := tx.Commit(); err != nil {
                return err
        }

        // no rollback to be performed
        tx = nil

        return nil
}

func (db *IdentityStore) Close() error {
        return common2.Close(db.readDB, db.writeDB)
}

func (db *IdentityStore) GetSchema() string {
        return fmt.Sprintf(`
                -- IdentityConfigurations
                CREATE TABLE IF NOT EXISTS %s (
                        id TEXT NOT NULL,
            type TEXT NOT NULL,  
                        url TEXT NOT NULL,
                        conf BYTEA,
                        raw BYTEA,
                        PRIMARY KEY(id, type, url)
                );
                CREATE INDEX IF NOT EXISTS idx_ic_type_%s ON %s ( type );
                CREATE INDEX IF NOT EXISTS idx_ic_id_type_%s ON %s ( id, type, url );

                -- IdentityInfo
                CREATE TABLE IF NOT EXISTS %s (
            identity_hash TEXT NOT NULL PRIMARY KEY,
                        identity BYTEA NOT NULL,
                        identity_audit_info BYTEA NOT NULL,
                        token_metadata BYTEA,
                        token_metadata_audit_info BYTEA
                );
                CREATE INDEX IF NOT EXISTS idx_audits_%s ON %s ( identity_hash );

                -- Signers
                CREATE TABLE IF NOT EXISTS %s (
            identity_hash TEXT NOT NULL PRIMARY KEY,
                        identity BYTEA NOT NULL,
                        info BYTEA
                );
                CREATE INDEX IF NOT EXISTS idx_signers_%s ON %s ( identity_hash );
                `,
                db.table.IdentityConfigurations,
                db.table.IdentityConfigurations, db.table.IdentityConfigurations,
                db.table.IdentityConfigurations, db.table.IdentityConfigurations,
                db.table.IdentityInfo,
                db.table.IdentityInfo, db.table.IdentityInfo,
                db.table.Signers,
                db.table.Signers, db.table.Signers,
        )
}

func (db *IdentityStore) storeSignerInfo(ctx context.Context, tx dbTransaction, h string, id tdriver.Identity, info []byte, updateCache bool) (bool, error) {
        logger.DebugfContext(ctx, "store signer info for [%s]", h)
        query, args := q.InsertInto(db.table.Signers).
                Fields("identity_hash", "identity", "info").
                Row(h, id, info).
                Format()

        logging.Debug(logger, query, h, utils.Hashable(info))
        exists := false
        _, err := tx.ExecContext(ctx, query, args...)
        if err != nil {
                if errors.Is(db.errorWrapper.WrapError(err), driver2.UniqueKeyViolation) {
                        logger.DebugfContext(ctx, "signer info [%s] exists, no error to return", h)
                        exists = true
                } else {
                        return exists, err
                }
        }
        if updateCache {
                db.signerInfoCache.Add(h, true)
        }
        logger.DebugfContext(ctx, "store signer info done")

        return exists, nil
}

func (db *IdentityStore) storeIdentityData(ctx context.Context, tx dbTransaction, h string, id []byte, identityAudit []byte, tokenMetadata []byte, tokenMetadataAudit []byte, updateCache bool) error {
        logger.DebugfContext(ctx, "store identity data for [%s]", h)
        query, args := q.InsertInto(db.table.IdentityInfo).
                Fields("identity_hash", "identity", "identity_audit_info", "token_metadata", "token_metadata_audit_info").
                Row(h, id, identityAudit, tokenMetadata, tokenMetadataAudit).
                Format()
        logging.Debug(logger, query, args)

        _, err := tx.ExecContext(ctx, query, args...)
        if err != nil {
                if !errors.Is(db.errorWrapper.WrapError(err), driver2.UniqueKeyViolation) {
                        return err
                }
                logger.DebugfContext(ctx, "identity data [%s] exists, no error to return", h)
        }

        if updateCache {
                logger.DebugfContext(ctx, "audit info cache update")
                db.auditInfoCache.Add(h, identityAudit)
                logger.DebugfContext(ctx, "audit info cache update done")
        }
        logger.DebugfContext(ctx, "store identity data for [%s] done", h)

        return nil
}

1	/*
2	Copyright IBM Corp. All Rights Reserved.
3
4	SPDX-License-Identifier: Apache-2.0
5	*/
6
7	package common
8
9	import (
10	"context"
11	"database/sql"
12	"fmt"
13
14	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
15	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/cache/secondcache"
16	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators"
17	driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver"
18	common2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/common"
19	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/common"
20	q "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query"
21	common3 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query/common"
22	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/query/cond"
23	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
24	"github.com/hyperledger-labs/fabric-token-sdk/token"
25	tdriver "github.com/hyperledger-labs/fabric-token-sdk/token/driver"
26	idriver "github.com/hyperledger-labs/fabric-token-sdk/token/services/identity/driver"
27	"github.com/hyperledger-labs/fabric-token-sdk/token/services/logging"
28	"github.com/hyperledger-labs/fabric-token-sdk/token/services/storage"
29	"github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/db/driver"
30	"github.com/hyperledger-labs/fabric-token-sdk/token/services/utils"
31	cache2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/cache"
32	)
33
34	type cache[T any] interface {
35	Get(key string) (T, bool)
36	GetOrLoad(key string, loader func() (T, error)) (T, bool, error)
37	Add(key string, value T)
38	Delete(key string)
39	}
40
41	type dbTransaction interface {
42	ExecContext(ctx context.Context, query string, args ...common3.Param) (sql.Result, error)
43	}
44
45	type identityTables struct {
46	IdentityConfigurations string
47	IdentityInfo string
48	Signers string
49	}
50
51	// IdentityStore is a SQL-backed implementation of the IdentityStore interface.
52	type IdentityStore struct {
53	readDB *sql.DB
54	writeDB *sql.DB
55	table identityTables
56	ci common3.CondInterpreter
57	notifier idriver.IdentityConfigurationNotifier
58
59	signerInfoCache cache[bool]
60	auditInfoCache cache[[]byte]
61	errorWrapper driver2.SQLErrorWrapper
62	}
63
64	func newIdentityStore(
65	readDB, writeDB *sql.DB,
66	tables identityTables,
67	singerInfoCache cache[bool],
68	auditInfoCache cache[[]byte],
69	ci common3.CondInterpreter,
70	errorWrapper driver2.SQLErrorWrapper,
71	notifier idriver.IdentityConfigurationNotifier,
72	) *IdentityStore {	47✔
73	return &IdentityStore{	47✔
74	readDB: readDB,	47✔
75	writeDB: writeDB,	47✔
76	table: tables,	47✔
77	signerInfoCache: singerInfoCache,	47✔
78	auditInfoCache: auditInfoCache,	47✔
79	ci: ci,	47✔
80	errorWrapper: errorWrapper,	47✔
81	notifier: notifier,	47✔
82	}	47✔
83	}	47✔
84
85	func NewCachedIdentityStore(
86	readDB, writeDB *sql.DB,
87	tables TableNames,
88	ci common3.CondInterpreter,
89	errorWrapper driver2.SQLErrorWrapper,
90	) (*IdentityStore, error) {	×
91	return NewIdentityStore(	×
92	readDB,	×
93	writeDB,	×
94	tables,	×
95	secondcache.NewTyped[bool](5000),	×
96	secondcache.NewTyped[[]byte](5000),	×
97	ci,	×
98	errorWrapper,	×
99	)	×
100	}	×
101
102	func NewNoCacheIdentityStore(
103	readDB, writeDB *sql.DB,
104	tables TableNames,
105	ci common3.CondInterpreter,
106	errorWrapper driver2.SQLErrorWrapper,
107	) (*IdentityStore, error) {	×
108	return NewIdentityStore(	×
109	readDB,	×
110	writeDB,	×
111	tables,	×
112	cache2.NewNoCache[bool](),	×
113	cache2.NewNoCache[[]byte](),	×
114	ci,	×
115	errorWrapper,	×
116	)	×
117	}	×
118
119	func NewIdentityStore(
120	readDB, writeDB *sql.DB,
121	tables TableNames,
122	signerInfoCache cache[bool],
123	auditInfoCache cache[[]byte],
124	ci common3.CondInterpreter,
125	errorWrapper driver2.SQLErrorWrapper,
126	) (*IdentityStore, error) {	×
127	return newIdentityStore(	×
128	readDB,	×
129	writeDB,	×
130	identityTables{	×
131	IdentityConfigurations: tables.IdentityConfigurations,	×
132	IdentityInfo: tables.IdentityInfo,	×
133	Signers: tables.Signers,	×
134	},	×
135	signerInfoCache,	×
136	auditInfoCache,	×
137	ci,	×
138	errorWrapper,	×
139	nil,	×
140	), nil	×
141	}	×
142
143	// NewIdentityStoreWithNotifier creates a new IdentityStore with a notifier.
144	func NewIdentityStoreWithNotifier(
145	readDB, writeDB *sql.DB,
146	tables TableNames,
147	signerInfoCache cache[bool],
148	auditInfoCache cache[[]byte],
149	ci common3.CondInterpreter,
150	errorWrapper driver2.SQLErrorWrapper,
151	notifier idriver.IdentityConfigurationNotifier,
152	) (*IdentityStore, error) {	47✔
153	return newIdentityStore(	47✔
154	readDB,	47✔
155	writeDB,	47✔
156	identityTables{	47✔
157	IdentityConfigurations: tables.IdentityConfigurations,	47✔
158	IdentityInfo: tables.IdentityInfo,	47✔
159	Signers: tables.Signers,	47✔
160	},	47✔
161	signerInfoCache,	47✔
162	auditInfoCache,	47✔
163	ci,	47✔
164	errorWrapper,	47✔
165	notifier,	47✔
166	), nil	47✔
167	}	47✔
168
169	func (db *IdentityStore) CreateSchema() error {	47✔
170	return common.InitSchema(db.writeDB, []string{db.GetSchema()}...)	47✔
171	}	47✔
172
173	// AddConfiguration stores an identity configuration in the database.
174	// It also enqueues an event to the notifier if available.
175	func (db *IdentityStore) AddConfiguration(ctx context.Context, wp driver.IdentityConfiguration) error {	45✔
176	query, args := q.InsertInto(db.table.IdentityConfigurations).	45✔
177	Fields("id", "type", "url", "conf", "raw").	45✔
178	Row(wp.ID, wp.Type, wp.URL, wp.Config, wp.Raw).	45✔
179	Format()	45✔
180	logging.Debug(logger, query, args)	45✔
181		45✔
182	_, err := db.writeDB.ExecContext(ctx, query, args...)	45✔
183	if err != nil {	45✔
184	return err	×
185	}	×
186
187	return nil	45✔
188	}
189
190	// GetConfiguration returns the configuration with the given id, type, and url.
191	func (db IdentityStore) GetConfiguration(ctx context.Context, id, typ, url string) (driver.IdentityConfiguration, error) {	10✔
192	query, args := q.Select().	10✔
193	FieldsByName("id", "type", "url", "conf", "raw").	10✔
194	From(q.Table(db.table.IdentityConfigurations)).	10✔
195	Where(cond.And(cond.Eq("id", id), cond.Eq("type", typ), cond.Eq("url", url))).	10✔
196	Format(db.ci)	10✔
197	logging.Debug(logger, query, args)	10✔
198	row := db.readDB.QueryRowContext(ctx, query, args...)	10✔
199	c := &driver.IdentityConfiguration{}	10✔
200	err := row.Scan(&c.ID, &c.Type, &c.URL, &c.Config, &c.Raw)	10✔
201	if err != nil {	16✔
202	if errors.Is(err, sql.ErrNoRows) {	12✔
203	return nil, nil	6✔
204	}	6✔
205
206	return nil, err	×
207	}
208
209	return c, nil	10✔
210	}
211
212	// IteratorConfigurations returns an iterator to all configurations of the given type.
213	func (db *IdentityStore) IteratorConfigurations(ctx context.Context, configurationType string) (idriver.IdentityConfigurationIterator, error) {	45✔
214	query, args := q.Select().	45✔
215	FieldsByName("id", "url", "conf", "raw").	45✔
216	From(q.Table(db.table.IdentityConfigurations)).	45✔
217	Where(cond.Eq("type", configurationType)).	45✔
218	Format(db.ci)	45✔
219	logging.Debug(logger, query, args)	45✔
220	rows, err := db.readDB.QueryContext(ctx, query, args...)	45✔
221	if err != nil {	51✔
222	return nil, err	6✔
223	}	6✔
224
225	return common.NewIterator(rows, func(c *driver.IdentityConfiguration) error {	90✔
226	c.Type = configurationType	45✔
227		45✔
228	return rows.Scan(&c.ID, &c.URL, &c.Config, &c.Raw)	45✔
229	}), nil	45✔
230	}
231
232	func (db *IdentityStore) ConfigurationExists(ctx context.Context, id, typ, url string) (bool, error) {	45✔
233	query, args := q.Select().	45✔
234	FieldsByName("id").	45✔
235	From(q.Table(db.table.IdentityConfigurations)).	45✔
236	Where(cond.And(cond.Eq("id", id), cond.Eq("type", typ), cond.Eq("url", url))).	45✔
237	Format(db.ci)	45✔
238	result, err := common.QueryUnique[string](db.readDB, query, args...)	45✔
239	if err != nil {	45✔
240	return false, errors.Wrapf(err, "failed getting configuration for [%s:%s:%s]", id, typ, url)	×
241	}	×
242	logger.DebugfContext(ctx, "found configuration for [%s:%s:%s]", id, typ, url)	45✔
243		45✔
244	return len(result) != 0, nil	45✔
245	}
246
247	// Notifier returns the IdentityNotifier associated with this store.
248	func (db *IdentityStore) Notifier() (idriver.IdentityConfigurationNotifier, error) {	45✔
249	if db.notifier == nil {	87✔
250	return nil, storage.ErrNotSupported	42✔
251	}	42✔
252
253	return db.notifier, nil	40✔
254	}
255
256	func (db *IdentityStore) StoreIdentityData(ctx context.Context, id []byte, identityAudit []byte, tokenMetadata []byte, tokenMetadataAudit []byte) error {	44✔
257	return db.storeIdentityData(ctx, db.writeDB, tdriver.Identity(id).UniqueID(), id, identityAudit, tokenMetadata, tokenMetadataAudit, true)	44✔
258	}	44✔
259
260	func (db *IdentityStore) GetAuditInfo(ctx context.Context, id []byte) ([]byte, error) {	45✔
261	h := token.Identity(id).String()	45✔
262	logger.DebugfContext(ctx, "get audit info for [%s]", h)	45✔
263		45✔
264	value, _, err := db.auditInfoCache.GetOrLoad(h, func() ([]byte, error) {	79✔
265	logger.DebugfContext(ctx, "load from backend identity data for [%s]", view.Identity(id))	34✔
266	query, args := q.Select().	34✔
267	FieldsByName("identity_audit_info").	34✔
268	From(q.Table(db.table.IdentityInfo)).	34✔
269	Where(cond.Eq("identity_hash", h)).	34✔
270	Format(db.ci)	34✔
271		34✔
272	return common.QueryUnique[[]byte](db.readDB, query, args...)	34✔
273	})	34✔
274
275	return value, err	45✔
276	}
277
278	func (db *IdentityStore) GetTokenInfo(ctx context.Context, id []byte) ([]byte, []byte, error) {	6✔
279	h := token.Identity(id).String()	6✔
280	logger.DebugfContext(ctx, "get identity data for [%s]", h)	6✔
281		6✔
282	query, args := q.Select().	6✔
283	FieldsByName("token_metadata", "token_metadata_audit_info").	6✔
284	From(q.Table(db.table.IdentityInfo)).	6✔
285	Where(cond.Eq("identity_hash", h)).	6✔
286	Format(db.ci)	6✔
287	logging.Debug(logger, query, args)	6✔
288		6✔
289	row := db.readDB.QueryRowContext(ctx, query, args...)	6✔
290	var tokenMetadata []byte	6✔
291	var tokenMetadataAuditInfo []byte	6✔
292	err := row.Scan(&tokenMetadata, &tokenMetadataAuditInfo)	6✔
293	if err != nil {	6✔
294	if errors.Is(err, sql.ErrNoRows) {	×
295	return nil, nil, nil	×
296	}	×
297
298	return nil, nil, errors.Wrapf(err, "error querying db")	×
299	}
300
301	return tokenMetadata, tokenMetadataAuditInfo, nil	6✔
302	}
303
304	func (db *IdentityStore) StoreSignerInfo(ctx context.Context, id tdriver.Identity, info []byte) error {	23✔
305	_, err := db.storeSignerInfo(ctx, db.writeDB, id.UniqueID(), id, info, true)	23✔
306		23✔
307	return err	23✔
308	}	23✔
309
310	func (db *IdentityStore) GetExistingSignerInfo(ctx context.Context, ids ...tdriver.Identity) ([]string, error) {	45✔
311	idHashes := make([]string, len(ids))	45✔
312	for i, id := range ids {	90✔
313	idHashes[i] = id.UniqueID()	45✔
314	}	45✔
315
316	result := make([]string, 0)	45✔
317	notFound := make([]string, 0)	45✔
318		45✔
319	for _, idHash := range idHashes {	90✔
320	if v, ok := db.signerInfoCache.Get(idHash); !ok {	90✔
321	notFound = append(notFound, idHash)	45✔
322	} else if v {	104✔
323	result = append(result, idHash)	15✔
324	}	15✔
325	}
326	if len(notFound) == 0 {	89✔
327	return result, nil	44✔
328	}	44✔
329
330	idHashes = notFound	45✔
331	notFound = make([]string, 0)	45✔
332	for _, idHash := range idHashes {	90✔
333	if v, ok := db.signerInfoCache.Get(idHash); !ok {	90✔
334	notFound = append(notFound, idHash)	45✔
335	} else if v {	45✔
336	result = append(result, idHash)	×
337	}	×
338	}
339	if len(notFound) == 0 {	45✔
340	return result, nil	×
341	}	×
342
343	idHashes = notFound	45✔
344		45✔
345	query, args := q.Select().	45✔
346	FieldsByName("identity_hash").	45✔
347	From(q.Table(db.table.Signers)).	45✔
348	Where(cond.In("identity_hash", idHashes...)).	45✔
349	Format(db.ci)	45✔
350		45✔
351	logging.Debug(logger, query, args)	45✔
352	rows, err := db.readDB.QueryContext(ctx, query, args...)	45✔
353	if err != nil {	45✔
354	return nil, errors.Wrapf(err, "error querying db")	×
355	}	×
356	it := common.NewIterator(rows, func(idHash *string) error { return rows.Scan(idHash) })	52✔
357
358	found, err := iterators.Reduce(it, iterators.ToSet[string]())	45✔
359	if err != nil {	45✔
360	return nil, err	×
361	}	×
362	for _, idHash := range idHashes {	90✔
363	db.signerInfoCache.Add(idHash, found.Contains(idHash))	45✔
364	}	45✔
365
366	return append(result, found.ToSlice()...), nil	45✔
367	}
368
369	func (db *IdentityStore) SignerInfoExists(ctx context.Context, id []byte) (bool, error) {	6✔
370	existing, err := db.GetExistingSignerInfo(ctx, id)	6✔
371	if err != nil {	6✔
372	return false, err	×
373	}	×
374
375	return len(existing) > 0, nil	6✔
376	}
377
378	func (db *IdentityStore) GetSignerInfo(ctx context.Context, identity []byte) ([]byte, error) {	26✔
379	query, args := q.Select().	26✔
380	FieldsByName("info").	26✔
381	From(q.Table(db.table.Signers)).	26✔
382	Where(cond.Eq("identity_hash", token.Identity(identity).UniqueID())).	26✔
383	Format(db.ci)	26✔
384		26✔
385	return common.QueryUniqueContext[[]byte](ctx, db.readDB, query, args...)	26✔
386	}	26✔
387
388	func (db IdentityStore) RegisterIdentityDescriptor(ctx context.Context, descriptor idriver.IdentityDescriptor, alias tdriver.Identity) error {	45✔
389	// store	45✔
390	logger.DebugfContext(ctx, "register identity descriptor...")	45✔
391	if err := db.registerIdentityDescriptor(ctx, descriptor, alias); err != nil {	45✔
392	logger.ErrorfContext(ctx, "register identity descriptor...failed: %v", err)	×
393		×
394	return err	×
395	}	×
396	logger.DebugfContext(ctx, "register identity descriptor...done")	45✔
397		45✔
398	// update all caches	45✔
399	logger.DebugfContext(ctx, "register identity descriptor...update caches...")	45✔
400	h := descriptor.Identity.UniqueID()	45✔
401	db.signerInfoCache.Add(h, true)	45✔
402	if len(descriptor.AuditInfo) != 0 {	90✔
403	db.auditInfoCache.Add(h, descriptor.AuditInfo)	45✔
404	}	45✔
405	if !alias.IsNone() && !descriptor.Identity.Equal(alias) {	90✔
406	h = alias.UniqueID()	45✔
407	db.signerInfoCache.Add(h, true)	45✔
408	if len(descriptor.AuditInfo) != 0 {	90✔
409	db.auditInfoCache.Add(h, descriptor.AuditInfo)	45✔
410	}	45✔
411	}
412	logger.DebugfContext(ctx, "register identity descriptor...update caches...done")	45✔
413		45✔
414	return nil	45✔
415	}
416
417	func (db *IdentityStore) registerIdentityDescriptor(
418	ctx context.Context,
419	descriptor *idriver.IdentityDescriptor,
420	alias tdriver.Identity,
421	) error {	45✔
422	if descriptor == nil {	45✔
423	return errors.New("identity descriptor is nil")	×
424	}	×
425	tx, err := db.writeDB.BeginTx(ctx, nil)	45✔
426	if err != nil {	45✔
427	return err	×
428	}	×
429	defer func() {	90✔
430	if tx != nil {	77✔
431	if err := tx.Rollback(); err != nil {	32✔
432	logger.ErrorfContext(ctx, "failed closing connection: %s", err)	×
433	}	×
434	}
435	}()
436
437	h := descriptor.Identity.UniqueID()	45✔
438		45✔
439	exists, err := db.storeSignerInfo(ctx, tx, h, descriptor.Identity, descriptor.SignerInfo, false)	45✔
440	if err != nil {	45✔
441	return errors.Wrapf(err, "failed to store signer info for descriptor's identity")	×
442	}	×
443	if exists {	77✔
444	// no need to continue	32✔
445	return nil	32✔
446	}	32✔
447
448	if len(descriptor.AuditInfo) != 0 {	90✔
449	err = db.storeIdentityData(ctx, tx, h, descriptor.Identity, descriptor.AuditInfo, nil, nil, false)	45✔
450	if err != nil {	45✔
451	return errors.Wrapf(err, "failed to store audit info for descriptor's identity")	×
452	}	×
453	}
454
455	if !alias.IsNone() && !descriptor.Identity.Equal(alias) {	90✔
456	h = alias.UniqueID()	45✔
457	_, err = db.storeSignerInfo(ctx, tx, h, alias, descriptor.SignerInfo, false)	45✔
458	if err != nil {	45✔
459	return errors.Wrapf(err, "failed to store signer info for alias")	×
460	}	×
461	if len(descriptor.AuditInfo) != 0 {	90✔
462	err = db.storeIdentityData(ctx, tx, h, alias, descriptor.AuditInfo, nil, nil, false)	45✔
463	if err != nil {	45✔
464	return errors.Wrapf(err, "failed to store audit info for alias")	×
465	}	×
466	}
467	}
468	if err := tx.Commit(); err != nil {	45✔
469	return err	×
470	}	×
471
472	// no rollback to be performed
473	tx = nil	45✔
474		45✔
475	return nil	45✔
476	}
477
478	func (db *IdentityStore) Close() error {	×
479	return common2.Close(db.readDB, db.writeDB)	×
480	}	×
481
482	func (db *IdentityStore) GetSchema() string {	47✔
483	return fmt.Sprintf(`	47✔
484	-- IdentityConfigurations	47✔
485	CREATE TABLE IF NOT EXISTS %s (	47✔
486	id TEXT NOT NULL,	47✔
487	type TEXT NOT NULL,	47✔
488	url TEXT NOT NULL,	47✔
489	conf BYTEA,	47✔
490	raw BYTEA,	47✔
491	PRIMARY KEY(id, type, url)	47✔
492	);	47✔
493	CREATE INDEX IF NOT EXISTS idx_ic_type_%s ON %s ( type );	47✔
494	CREATE INDEX IF NOT EXISTS idx_ic_id_type_%s ON %s ( id, type, url );	47✔
495		47✔
496	-- IdentityInfo	47✔
497	CREATE TABLE IF NOT EXISTS %s (	47✔
498	identity_hash TEXT NOT NULL PRIMARY KEY,	47✔
499	identity BYTEA NOT NULL,	47✔
500	identity_audit_info BYTEA NOT NULL,	47✔
501	token_metadata BYTEA,	47✔
502	token_metadata_audit_info BYTEA	47✔
503	);	47✔
504	CREATE INDEX IF NOT EXISTS idx_audits_%s ON %s ( identity_hash );	47✔
505		47✔
506	-- Signers	47✔
507	CREATE TABLE IF NOT EXISTS %s (	47✔
508	identity_hash TEXT NOT NULL PRIMARY KEY,	47✔
509	identity BYTEA NOT NULL,	47✔
510	info BYTEA	47✔
511	);	47✔
512	CREATE INDEX IF NOT EXISTS idx_signers_%s ON %s ( identity_hash );	47✔
513	`,	47✔
514	db.table.IdentityConfigurations,	47✔
515	db.table.IdentityConfigurations, db.table.IdentityConfigurations,	47✔
516	db.table.IdentityConfigurations, db.table.IdentityConfigurations,	47✔
517	db.table.IdentityInfo,	47✔
518	db.table.IdentityInfo, db.table.IdentityInfo,	47✔
519	db.table.Signers,	47✔
520	db.table.Signers, db.table.Signers,	47✔
521	)	47✔
522	}	47✔
523
524	func (db *IdentityStore) storeSignerInfo(ctx context.Context, tx dbTransaction, h string, id tdriver.Identity, info []byte, updateCache bool) (bool, error) {	45✔
525	logger.DebugfContext(ctx, "store signer info for [%s]", h)	45✔
526	query, args := q.InsertInto(db.table.Signers).	45✔
527	Fields("identity_hash", "identity", "info").	45✔
528	Row(h, id, info).	45✔
529	Format()	45✔
530		45✔
531	logging.Debug(logger, query, h, utils.Hashable(info))	45✔
532	exists := false	45✔
533	_, err := tx.ExecContext(ctx, query, args...)	45✔
534	if err != nil {	84✔
535	if errors.Is(db.errorWrapper.WrapError(err), driver2.UniqueKeyViolation) {	78✔
536	logger.DebugfContext(ctx, "signer info [%s] exists, no error to return", h)	39✔
537	exists = true	39✔
538	} else {	39✔
539	return exists, err	×
540	}	×
541	}
542	if updateCache {	68✔
543	db.signerInfoCache.Add(h, true)	23✔
544	}	23✔
545	logger.DebugfContext(ctx, "store signer info done")	45✔
546		45✔
547	return exists, nil	45✔
548	}
549
550	func (db *IdentityStore) storeIdentityData(ctx context.Context, tx dbTransaction, h string, id []byte, identityAudit []byte, tokenMetadata []byte, tokenMetadataAudit []byte, updateCache bool) error {	45✔
551	logger.DebugfContext(ctx, "store identity data for [%s]", h)	45✔
552	query, args := q.InsertInto(db.table.IdentityInfo).	45✔
553	Fields("identity_hash", "identity", "identity_audit_info", "token_metadata", "token_metadata_audit_info").	45✔
554	Row(h, id, identityAudit, tokenMetadata, tokenMetadataAudit).	45✔
555	Format()	45✔
556	logging.Debug(logger, query, args)	45✔
557		45✔
558	_, err := tx.ExecContext(ctx, query, args...)	45✔
559	if err != nil {	57✔
560	if !errors.Is(db.errorWrapper.WrapError(err), driver2.UniqueKeyViolation) {	12✔
561	return err	×
562	}	×
563	logger.DebugfContext(ctx, "identity data [%s] exists, no error to return", h)	12✔
564	}
565
566	if updateCache {	89✔
567	logger.DebugfContext(ctx, "audit info cache update")	44✔
568	db.auditInfoCache.Add(h, identityAudit)	44✔
569	logger.DebugfContext(ctx, "audit info cache update done")	44✔
570	}	44✔
571	logger.DebugfContext(ctx, "store identity data for [%s] done", h)	45✔
572		45✔
573	return nil	45✔
574	}

hyperledger-labs / fabric-token-sdk / 24770254403

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous