audius L1 (#9264)

core/.dockerignore

@@ -0,0 +1,3 @@
+**/*.env
+*.env
+/tmp

core/.gitignore

@@ -0,0 +1 @@
+**/tmp

core/Makefile

@@ -0,0 +1,34 @@
+dev-discovery-1::
+	@go run main.go -env-file=./infra/dev_config/discovery-one.env
+
+dev-content-1::
+	@go run main.go -env-file=./infra/dev_config/content-one.env
+
+dev-content-2::
+	@go run main.go -env-file=./infra/dev_config/content-two.env
+
+dev-content-3::
+	@go run main.go -env-file=./infra/dev_config/content-three.env
+
+up:
+	make down
+	make gen
+	cd infra && docker compose up --build -d
+
+down:
+	cd infra && docker compose down
+
+deps::
+	@go install github.com/onsi/ginkgo/v2/[email protected]
+	@brew install protobuf
+	@go install github.com/sqlc-dev/sqlc/cmd/sqlc@latest
+	@go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
+	@go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
+
+gen::
+	cd db && sqlc generate
+	go mod tidy
+
+test::
+	@go test -v ./...
+	@cd test/integration && ginkgo -r

core/README.md

@@ -0,0 +1,54 @@
+# audius core
+
+The distributed event log that binds discovery, content, and identity together. The audius L1.
+
+## architecture
+
+Built with cometbft, audius core is the L1 blockchain that stores all the canon events in the audius protocol. The core is the library that allows you to build node types around this source, publish events, index data, and materialize views that then power audius applications.
+
+```mermaid
+graph TD
+    L1["Audius L1"]
+    L2a["Discovery Provider"]
+    L2b["Content Node"]
+    L2c["Identity Service"]
+
+    L1 --> |"indexes discovery\n related data"| L2a
+    L1 --> |"indexes content\n related data"| L2b
+    L1 --> |"indexes identity\n related data"| L2c
+
+```
+
+## configuration
+
+Whether running as an independent image or embedded, core requires some environment variables to run.
+
+### local cluster
+
+### running in production
+
+
+## testing
+
+To run tests simply run the make command. This will run both unit and integration tests.
+
+```bash
+make test
+```
+
+### unit tests
+
+Place unit tests next to appropriate files in the standard go fashion
+
+```bash
+core/signature.go
+core/signature_test.go
+```
+
+### integration tests
+
+Integration tests are written with [ginkgo](https://github.com/onsi/ginkgo) and [gomega](https://github.com/onsi/gomega) to be more readable, the way to generate one is to use the ginkgo cli installed in the `make deps` command.
+
+```bash
+ginkgo generate NewTestName
+```

core/accounts/signature.go

@@ -0,0 +1 @@
+package accounts

core/accounts/wallet.go

@@ -0,0 +1,24 @@
+package accounts
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+
+	"github.com/cometbft/cometbft/crypto/ed25519"
+)
+
+func EthToCometKey(hexPkey string) (ed25519.PrivKey, error) {
+	ethPkeyBytes, err := hex.DecodeString(hexPkey)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(ethPkeyBytes) != 32 {
+		return nil, errors.New("private key length not 32")
+	}
+
+	hash := sha256.Sum256(ethPkeyBytes)
+	eckey := ed25519.GenPrivKeyFromSecret(hash[:])
+	return eckey, nil
+}

core/accounts/wallet_test.go

@@ -0,0 +1 @@
+package accounts

core/chain/abci.go

@@ -0,0 +1,170 @@
+package chain
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+
+	"github.com/AudiusProject/audius-protocol/core/common"
+	"github.com/AudiusProject/audius-protocol/core/db"
+	abcitypes "github.com/cometbft/cometbft/abci/types"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type KVStoreApplication struct {
+	logger       *common.Logger
+	queries      *db.Queries
+	pool         *pgxpool.Pool
+	onGoingBlock pgx.Tx
+}
+
+var _ abcitypes.Application = (*KVStoreApplication)(nil)
+
+func NewKVStoreApplication(logger *common.Logger, pool *pgxpool.Pool) *KVStoreApplication {
+	return &KVStoreApplication{
+		logger:       logger,
+		queries:      db.New(pool),
+		pool:         pool,
+		onGoingBlock: nil,
+	}
+}
+
+func (app *KVStoreApplication) Info(_ context.Context, info *abcitypes.InfoRequest) (*abcitypes.InfoResponse, error) {
+	return &abcitypes.InfoResponse{}, nil
+}
+
+func (app *KVStoreApplication) Query(ctx context.Context, req *abcitypes.QueryRequest) (*abcitypes.QueryResponse, error) {
+	resp := abcitypes.QueryResponse{Key: req.Data}
+
+	kv, err := app.queries.GetKey(ctx, string(req.Data))
+	if err != nil {
+		resp.Log = err.Error()
+		return &resp, err
+	}
+
+	value := []byte(kv.Value)
+	resp.Log = "exists"
+	resp.Value = value
+
+	return &resp, nil
+}
+
+func (app *KVStoreApplication) CheckTx(_ context.Context, check *abcitypes.CheckTxRequest) (*abcitypes.CheckTxResponse, error) {
+	code := app.isValid(check.Tx)
+	return &abcitypes.CheckTxResponse{Code: code}, nil
+}
+
+func (app *KVStoreApplication) InitChain(_ context.Context, chain *abcitypes.InitChainRequest) (*abcitypes.InitChainResponse, error) {
+	return &abcitypes.InitChainResponse{}, nil
+}
+
+func (app *KVStoreApplication) PrepareProposal(_ context.Context, proposal *abcitypes.PrepareProposalRequest) (*abcitypes.PrepareProposalResponse, error) {
+	return &abcitypes.PrepareProposalResponse{Txs: proposal.Txs}, nil
+}
+
+func (app *KVStoreApplication) ProcessProposal(_ context.Context, proposal *abcitypes.ProcessProposalRequest) (*abcitypes.ProcessProposalResponse, error) {
+	return &abcitypes.ProcessProposalResponse{Status: abcitypes.PROCESS_PROPOSAL_STATUS_ACCEPT}, nil
+}
+
+func (app *KVStoreApplication) FinalizeBlock(ctx context.Context, req *abcitypes.FinalizeBlockRequest) (*abcitypes.FinalizeBlockResponse, error) {
+	logger := app.logger
+	var txs = make([]*abcitypes.ExecTxResult, len(req.Txs))
+
+	// early out if empty block
+	if len(txs) == 0 {
+		return &abcitypes.FinalizeBlockResponse{
+			TxResults: txs,
+		}, nil
+	}
+
+	// open in progres pg transaction
+	app.startInProgressTx(ctx)
+	for i, tx := range req.Txs {
+		if code := app.isValid(tx); code != 0 {
+			logger.Errorf("Error: invalid transaction index %v", i)
+			txs[i] = &abcitypes.ExecTxResult{Code: code}
+		} else {
+			parts := bytes.SplitN(tx, []byte("="), 2)
+			key, value := parts[0], parts[1]
+			logger.Infof("Adding key %s with value %s", key, value)
+
+			qtx := app.getDb()
+
+			hash := sha256.Sum256(tx)
+			txHash := hex.EncodeToString(hash[:])
+
+			params := db.InsertKVStoreParams{
+				Key:    string(key),
+				Value:  string(value),
+				TxHash: txHash,
+			}
+
+			record, err := qtx.InsertKVStore(ctx, params)
+			if err != nil {
+				logger.Errorf("failed to persisted kv entry %v", err)
+			}
+
+			txs[i] = &abcitypes.ExecTxResult{
+				Code: 0,
+				Events: []abcitypes.Event{
+					{
+						Type: "app",
+						Attributes: []abcitypes.EventAttribute{
+							{Key: "key", Value: record.Key, Index: true},
+							{Key: "value", Value: record.Value, Index: true},
+						},
+					},
+				},
+			}
+		}
+	}
+
+	return &abcitypes.FinalizeBlockResponse{
+		TxResults: txs,
+	}, nil
+}
+
+func (app KVStoreApplication) Commit(ctx context.Context, commit *abcitypes.CommitRequest) (*abcitypes.CommitResponse, error) {
+	app.logger.Info("in commit phase", "onGoingBlock", app.onGoingBlock)
+	if err := app.commitInProgressTx(ctx); err != nil {
+		app.logger.Error("failure to commit tx", "error", err)
+		return &abcitypes.CommitResponse{}, err
+	}
+	return &abcitypes.CommitResponse{}, nil
+}
+
+func (app *KVStoreApplication) ListSnapshots(_ context.Context, snapshots *abcitypes.ListSnapshotsRequest) (*abcitypes.ListSnapshotsResponse, error) {
+	return &abcitypes.ListSnapshotsResponse{}, nil
+}
+
+func (app *KVStoreApplication) OfferSnapshot(_ context.Context, snapshot *abcitypes.OfferSnapshotRequest) (*abcitypes.OfferSnapshotResponse, error) {
+	return &abcitypes.OfferSnapshotResponse{}, nil
+}
+
+func (app *KVStoreApplication) LoadSnapshotChunk(_ context.Context, chunk *abcitypes.LoadSnapshotChunkRequest) (*abcitypes.LoadSnapshotChunkResponse, error) {
+	return &abcitypes.LoadSnapshotChunkResponse{}, nil
+}
+
+func (app *KVStoreApplication) ApplySnapshotChunk(_ context.Context, chunk *abcitypes.ApplySnapshotChunkRequest) (*abcitypes.ApplySnapshotChunkResponse, error) {
+	return &abcitypes.ApplySnapshotChunkResponse{Result: abcitypes.APPLY_SNAPSHOT_CHUNK_RESULT_ACCEPT}, nil
+}
+
+func (app KVStoreApplication) ExtendVote(_ context.Context, extend *abcitypes.ExtendVoteRequest) (*abcitypes.ExtendVoteResponse, error) {
+	return &abcitypes.ExtendVoteResponse{}, nil
+}
+
+func (app *KVStoreApplication) VerifyVoteExtension(_ context.Context, verify *abcitypes.VerifyVoteExtensionRequest) (*abcitypes.VerifyVoteExtensionResponse, error) {
+	return &abcitypes.VerifyVoteExtensionResponse{}, nil
+}
+
+func (app *KVStoreApplication) isValid(tx []byte) uint32 {
+	// check format
+	parts := bytes.Split(tx, []byte("="))
+	if len(parts) != 2 {
+		return 1
+	}
+
+	return 0
+}

core/chain/db.go

@@ -0,0 +1,40 @@
+package chain
+
+import (
+	"context"
+	"errors"
+
+	"github.com/AudiusProject/audius-protocol/core/db"
+	"github.com/jackc/pgx/v5"
+)
+
+// returns in current postgres tx for this block
+func (c *KVStoreApplication) getDb() *db.Queries {
+	return c.queries.WithTx(c.onGoingBlock)
+}
+
+func (c *KVStoreApplication) startInProgressTx(ctx context.Context) error {
+	dbTx, err := c.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+
+	c.onGoingBlock = dbTx
+	return nil
+}
+
+// commits the current tx that's finished indexing
+func (c *KVStoreApplication) commitInProgressTx(ctx context.Context) error {
+	if c.onGoingBlock != nil {
+		err := c.onGoingBlock.Commit(ctx)
+		if err != nil {
+			if errors.Is(err, pgx.ErrTxClosed) {
+				c.onGoingBlock = nil
+				return nil
+			}
+			return err
+		}
+		c.onGoingBlock = nil
+	}
+	return nil
+}