Implement basic create command.

internal/handlers/handler.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"sync/atomic"
 
-	"github.com/jackc/pgx/v4"
 	"go.uber.org/zap"
 
 	"github.com/FerretDB/FerretDB/internal/handlers/common"
@@ -156,6 +155,8 @@ func (h *Handler) handleOpMsg(ctx context.Context, msg *wire.OpMsg) (*wire.OpMsg
 	switch cmd {
 	case "buildinfo":
 		return h.shared.MsgBuildInfo(ctx, msg)
+	case "create":
+		return h.shared.MsgCreate(ctx, msg)
 	case "drop":
 		return h.shared.MsgDrop(ctx, msg)
 	case "dropdatabase":
@@ -256,10 +257,8 @@ func (h *Handler) msgStorage(ctx context.Context, msg *wire.OpMsg) (common.Stora
 			return h.jsonb1, nil
 		}
 		if !tableExist {
-			sql = `CREATE TABLE ` + pgx.Identifier{db, collection}.Sanitize() + ` (_jsonb jsonb)`
-			_, err := h.pgPool.Exec(ctx, sql)
 			fields := []zap.Field{zap.String("schema", db), zap.String("table", collection)}
-			if err != nil {
+			if err := shared.CreateCollection(ctx, h.pgPool, db, collection); err != nil {
 				fields = append(fields, zap.Error(err))
 				h.l.Warn("Failed to create jsonb1 table.", fields...)
 			} else {

internal/handlers/handler_test.go

@@ -59,6 +59,22 @@ func setup(t *testing.T, poolOpts *testutil.PoolOpts) (context.Context, *Handler
 	return ctx, handler, pool
 }
 
+func createDB(ctx context.Context, pool *pg.Pool, name string) error {
+	_, err := pool.Exec(ctx, fmt.Sprintf(
+		`CREATE SCHEMA %s`,
+		pgx.Identifier{name}.Sanitize(),
+	))
+	return err
+}
+
+func dropDB(ctx context.Context, pool *pg.Pool, name string) error {
+	_, err := pool.Exec(ctx, fmt.Sprintf(
+		`DROP SCHEMA IF EXISTS %s CASCADE`,
+		pgx.Identifier{name}.Sanitize(),
+	))
+	return err
+}
+
 func TestDropDatabase(t *testing.T) { //nolint:paralleltest,tparallel // affects a global list of databases
 	ctx, handler, pool := setup(t, nil)
 
@@ -90,20 +106,11 @@ func TestDropDatabase(t *testing.T) { //nolint:paralleltest,tparallel // affects
 		},
 	}
 
-	_, err := pool.Exec(ctx, fmt.Sprintf(
-		`CREATE SCHEMA %s`,
-		pgx.Identifier{dummyDbName}.Sanitize(),
-	))
-	require.NoError(t, err)
 	dummyTableName := "dummy_table"
+	require.NoError(t, createDB(ctx, pool, dummyDbName))
+	require.NoError(t, shared.CreateCollection(ctx, pool, dummyDbName, dummyTableName))
 
-	_, err = pool.Exec(ctx, fmt.Sprintf(
-		`CREATE TABLE %s (_jsonb jsonb)`,
-		pgx.Identifier{dummyDbName, dummyTableName}.Sanitize(),
-	))
-	require.NoError(t, err)
-
-	_, err = pool.Exec(ctx, fmt.Sprintf(
+	_, err := pool.Exec(ctx, fmt.Sprintf(
 		`INSERT INTO %s(_jsonb) VALUES ('{"key": "value"}')`,
 		pgx.Identifier{dummyDbName, dummyTableName}.Sanitize(),
 	))
@@ -642,3 +649,78 @@ func TestReadOnlyHandlers(t *testing.T) {
 		})
 	}
 }
+
+func TestCreate(t *testing.T) { //nolint:paralleltest,tparallel // affects a global list of databases
+	ctx, handler, pool := setup(t, nil)
+
+	type testCase struct {
+		req     types.Document
+		resp    types.Document
+		success bool
+	}
+
+	dbName := "test_create_db"
+	collectionNew := "new_collection"
+	collectionExisted := "existed_collection"
+
+	testCases := map[string]testCase{
+		"new-collection": {
+			req: types.MustMakeDocument(
+				"create", collectionNew,
+				"$db", dbName,
+			),
+			success: true,
+			resp: types.MustMakeDocument(
+				"created", fmt.Sprintf("%s.%s", dbName, collectionNew),
+				"ok", float64(1),
+			),
+		},
+		"existed-collection": {
+			req: types.MustMakeDocument(
+				"create", collectionExisted,
+				"$db", dbName,
+			),
+			success: false,
+			resp:    types.MustMakeDocument(),
+		},
+	}
+
+	require.NoError(t, dropDB(ctx, pool, dbName))
+	require.NoError(t, createDB(ctx, pool, dbName))
+	require.NoError(t, shared.CreateCollection(ctx, pool, dbName, collectionExisted))
+
+	for name, tc := range testCases { //nolint:paralleltest // false positive
+		name, tc := name, tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			reqHeader := wire.MsgHeader{
+				RequestID: 1,
+				OpCode:    wire.OP_MSG,
+			}
+
+			var reqMsg wire.OpMsg
+			require.NoError(t, reqMsg.SetSections(wire.OpMsgSection{
+				Documents: []types.Document{tc.req},
+			}))
+
+			_, resBody, closeConn := handler.Handle(ctx, &reqHeader, &reqMsg)
+			if !tc.success {
+				require.True(t, closeConn, "%s", wire.DumpMsgBody(resBody))
+				return
+			}
+
+			require.False(t, closeConn, "%s", wire.DumpMsgBody(resBody))
+			actual, err := resBody.(*wire.OpMsg).Document()
+			require.NoError(t, err)
+
+			assert.Equal(t, actual.Map(), tc.resp.Map())
+
+			// checking tables
+			query := `SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = $1 and table_name = $2`
+			var count int
+			require.NoError(t, pool.QueryRow(ctx, query, dbName, collectionNew).Scan(&count))
+			assert.Equal(t, count, 1)
+		})
+	}
+}

internal/handlers/shared/msg_create.go

@@ -0,0 +1,62 @@
+// Copyright 2021 FerretDB Inc.
+//
+// Licensed 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 shared
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v4"
+
+	"github.com/FerretDB/FerretDB/internal/pg"
+	"github.com/FerretDB/FerretDB/internal/types"
+	"github.com/FerretDB/FerretDB/internal/util/lazyerrors"
+	"github.com/FerretDB/FerretDB/internal/wire"
+)
+
+// CreateCollection creates a new collection in db.
+func CreateCollection(ctx context.Context, pgPool *pg.Pool, db, collection string) error {
+	sql := `CREATE TABLE ` + pgx.Identifier{db, collection}.Sanitize() + ` (_jsonb jsonb)`
+	_, err := pgPool.Exec(ctx, sql)
+	return err
+}
+
+// MsgCreate adds a collection or view into the database.
+func (h *Handler) MsgCreate(ctx context.Context, msg *wire.OpMsg) (*wire.OpMsg, error) {
+	document, err := msg.Document()
+	if err != nil {
+		return nil, lazyerrors.Error(err)
+	}
+
+	m := document.Map()
+	collection := m[document.Command()].(string)
+	db := m["$db"].(string)
+
+	if err := CreateCollection(ctx, h.pgPool, db, collection); err != nil {
+		return nil, lazyerrors.Error(err)
+	}
+
+	var reply wire.OpMsg
+	err = reply.SetSections(wire.OpMsgSection{
+		Documents: []types.Document{types.MustMakeDocument(
+			"created", db+"."+collection,
+			"ok", float64(1),
+		)},
+	})
+	if err != nil {
+		return nil, lazyerrors.Error(err)
+	}
+
+	return &reply, nil
+}