1% Licensed under the Apache License, Version 2.0 (the "License"); you may not
2% use this file except in compliance with the License. You may obtain a copy of
3% the License at
4%
5%   http://www.apache.org/licenses/LICENSE-2.0
6%
7% Unless required by applicable law or agreed to in writing, software
8% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10% License for the specific language governing permissions and limitations under
11% the License.
12
13-module(couch_partition).
14
15
16-export([
17    extract/1,
18    from_docid/1,
19    is_member/2,
20
21    start_key/1,
22    end_key/1,
23    shard_key/1,
24
25    validate_dbname/2,
26    validate_docid/1,
27    validate_partition/1,
28
29    hash/1
30]).
31
32
33-include_lib("couch/include/couch_db.hrl").
34
35
36extract(Value) when is_binary(Value) ->
37    case binary:split(Value, <<":">>) of
38        [Partition, Rest] ->
39            {Partition, Rest};
40        _ ->
41            undefined
42    end;
43
44extract(_) ->
45    undefined.
46
47
48from_docid(DocId) ->
49    case extract(DocId) of
50        undefined ->
51            throw({illegal_docid, <<"Doc id must be of form partition:id">>});
52        {Partition, _} ->
53            Partition
54    end.
55
56
57is_member(DocId, Partition) ->
58    case extract(DocId) of
59        {Partition, _} ->
60            true;
61        _ ->
62            false
63    end.
64
65
66start_key(Partition) ->
67    <<Partition/binary, ":">>.
68
69
70end_key(Partition) ->
71    <<Partition/binary, ";">>.
72
73
74shard_key(Partition) ->
75    <<Partition/binary, ":foo">>.
76
77
78validate_dbname(DbName, Options) when is_list(DbName) ->
79    validate_dbname(?l2b(DbName), Options);
80validate_dbname(DbName, Options) when is_binary(DbName) ->
81    Props = couch_util:get_value(props, Options, []),
82    IsPartitioned = couch_util:get_value(partitioned, Props, false),
83
84    if not IsPartitioned -> ok; true ->
85
86        DbsDbName = config:get("mem3", "shards_db", "_dbs"),
87        NodesDbName = config:get("mem3", "nodes_db", "_nodes"),
88        UsersDbSuffix = config:get("couchdb", "users_db_suffix", "_users"),
89        Suffix = couch_db:dbname_suffix(DbName),
90
91        SysDbNames = [
92                iolist_to_binary(DbsDbName),
93                iolist_to_binary(NodesDbName)
94                | ?SYSTEM_DATABASES
95            ],
96
97        Suffices = [
98                <<"_replicator">>,
99                <<"_users">>,
100                iolist_to_binary(UsersDbSuffix)
101            ],
102
103        IsSysDb = lists:member(DbName, SysDbNames)
104                orelse lists:member(Suffix, Suffices),
105
106        if not IsSysDb -> ok; true ->
107            throw({bad_request, <<"Cannot partition a system database">>})
108        end
109    end.
110
111
112validate_docid(<<"_design/", _/binary>>) ->
113    ok;
114validate_docid(<<"_local/", _/binary>>) ->
115    ok;
116validate_docid(DocId) when is_binary(DocId) ->
117    % When this function is called we already know that
118    % DocId is already valid thus we only need to
119    % ensure that the partition exists and is not empty.
120    case extract(DocId) of
121        undefined ->
122            throw({illegal_docid, <<"Doc id must be of form partition:id">>});
123        {Partition, PartitionedDocId} ->
124            validate_partition(Partition),
125            couch_doc:validate_docid(PartitionedDocId)
126    end.
127
128
129validate_partition(<<>>) ->
130    throw({illegal_partition, <<"Partition must not be empty">>});
131validate_partition(Partition) when is_binary(Partition) ->
132    case Partition of
133        <<"_", _/binary>> ->
134            Msg1 = <<"Partition must not start with an underscore">>,
135            throw({illegal_partition, Msg1});
136        _ ->
137            ok
138    end,
139    case couch_util:validate_utf8(Partition) of
140        true ->
141            ok;
142        false ->
143            Msg2 = <<"Partition must be valid UTF-8">>,
144            throw({illegal_partition, Msg2})
145    end,
146    case extract(Partition) of
147        {_, _} ->
148            Msg3 = <<"Partition must not contain a colon">>,
149            throw({illegal_partition, Msg3});
150        undefined ->
151            ok
152    end;
153validate_partition(_) ->
154    throw({illegal_partition, <<"Partition must be a string">>}).
155
156
157% Document ids that start with an underscore
158% (i.e., _local and _design) do not contain a
159% partition and thus do not use the partition
160% hashing.
161hash(<<"_", _/binary>> = DocId) ->
162    erlang:crc32(DocId);
163hash(DocId) when is_binary(DocId) ->
164    erlang:crc32(from_docid(DocId)).
165