fix: State add flush and fix migration bug (#1039)

This fixes a bug in state backend where table weren't created.  

I also tested it with
cloudquery/cloudquery#11855 which seems to work
with minor modifications (will open a PR shortly to the CLI PR)

internal/clients/state/v3/state.go

@@ -3,7 +3,6 @@ package state
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"io"
 	"sync"
 
@@ -71,6 +70,9 @@ func NewClient(ctx context.Context, pbClient pb.PluginClient, tableName string)
 	}); err != nil {
 		return nil, err
 	}
+	if _, err := writeClient.CloseAndRecv(); err != nil {
+		return nil, err
+	}
 
 	readClient, err := c.client.Read(ctx, &pb.Read_Request{
 		Table: tableBytes,
@@ -158,5 +160,5 @@ func (c *Client) GetKey(_ context.Context, key string) (string, error) {
 	if val, ok := c.mem[key]; ok {
 		return val, nil
 	}
-	return "", fmt.Errorf("key not found")
+	return "", nil
 }

state/state.go

@@ -14,6 +14,7 @@ import (
 type Client interface {
 	SetKey(ctx context.Context, key string, value string) error
 	GetKey(ctx context.Context, key string) (string, error)
+	Flush(ctx context.Context) error
 }
 
 func NewClient(ctx context.Context, conn *grpc.ClientConn, tableName string) (Client, error) {